# Native Kafka clients

> A Kafka producer, consumer or tool of any library connects to the gateway with no code change. Available since 1.6.0. Product side: Native Kafka clients and…

Guide - Since version 1.6.0 - Flexway v1.6.3 - https://flexway.io/en/docs/developer-kafka/

> A Kafka producer, consumer or tool of any library connects to the gateway with no code change. Available since `1.6.0`.
>
> Product side: [Native Kafka clients and field encryption](https://flexway.io/en/docs/product-kafka/); operations side: [Native Kafka clients: a virtual cluster per tenant](https://flexway.io/en/docs/operations-kafka-clients/), [Field encryption](https://flexway.io/en/docs/operations-kafka-encryption/).

## Connecting

| Property | Value |
| --- | --- |
| `bootstrap.servers` | `bootstrap.<cluster>.<domain>:9093`, given by the operator |
| Encryption | TLS |
| Identity | `OAUTHBEARER` (a token of the declared issuer), `PLAIN` (the API key's id as user, the key as password) or a client certificate |
| `group.protocol` | `classic` (the default): KIP-848's protocol is not served |

```properties
# librdkafka (confluent-kafka-go, Python, .NET) - an OAUTHBEARER token
bootstrap.servers=bootstrap.shop.kafka.example.com:9093
security.protocol=SASL_SSL
sasl.mechanisms=OAUTHBEARER
sasl.oauthbearer.method=oidc
sasl.oauthbearer.client.id=billing
sasl.oauthbearer.client.secret=${BILLING_SECRET}
sasl.oauthbearer.token.endpoint.url=https://login.example.net/token
sasl.oauthbearer.scope=orders:read
group.id=billing
```

```properties
# Java client - an API key over PLAIN
bootstrap.servers=bootstrap.shop.kafka.example.com:9093
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="acme-orders-reader" password="${FLEXWAY_KAFKA_KEY}";
```

```go
// franz-go - an API key over PLAIN
cl, err := kgo.NewClient(
	kgo.SeedBrokers("bootstrap.shop.kafka.example.com:9093"),
	kgo.DialTLSConfig(&tls.Config{MinVersion: tls.VersionTLS12}),
	kgo.SASL(plain.Auth{User: "acme-orders-producer", Pass: os.Getenv("FLEXWAY_KAFKA_KEY")}.AsMechanism()),
)
```

## What you see

Your **virtual cluster**: only the topics the contract opens to you, under their short names. Your
groups and transactional ids are placed by the gateway in your tenant's namespace, without you
changing them. The gateway presents its own credentials to the broker.

| Kafka code | Meaning | What to do |
| --- | --- | --- |
| `TOPIC_AUTHORIZATION_FAILED`, `GROUP_AUTHORIZATION_FAILED`, `TRANSACTIONAL_ID_AUTHORIZATION_FAILED` | Operation not open to your identity | Ask for access |
| `UNKNOWN_TOPIC_OR_PARTITION` | Topic outside your virtual cluster | Check the short name |
| `SASL_AUTHENTICATION_FAILED` | Credential refused or revoked | Renew the key or the token |
| `POLICY_VIOLATION` (44) | Daily or monthly quota spent (on a write only: reads are not counted yet), or administration refused | Wait for the next window |
| `INVALID_RECORD` (87) | A `flexway-*` header you wrote, a value its schema does not decode | Fix the record |
| `KAFKA_STORAGE_ERROR` (56) | KMS unavailable: nothing is written in clear | The client retries on its own |
| `RESOURCE_NOT_FOUND` (91) | Encryption key destroyed | Do not retry |

A connection is closed when your credential is revoked or at the end of the session lifetime: the
client reconnects and re-authenticates on its own (KIP-368).

## Quotas: `throttle_time_ms`

Past your rate, the response arrives **at once** with `throttle_time_ms`, and the gateway stops reading
your connection for that long. A Java or franz-go client waits on its own; librdkafka ignores the
field but is slowed all the same. Keep `request.timeout.ms` (30 s by default) above the throttle cap
the operator announces.

## Encrypted fields

The fields the contract declares sensitive are encrypted by the gateway. A consumer whose token carries
the **decrypt scope** receives them in clear; others receive the encrypted value (base64 in a text
field). If the key was destroyed (a wanted erasure), the record arrives with the field **still
encrypted** and the header `flexway-encryption: key_destroyed` - read it before using the field:

```python
for msg in consumer.consume(100, timeout=1.0):
    headers = dict(msg.headers() or [])
    if headers.get("flexway-encryption") == b"key_destroyed":
        continue  # wanted erasure: do not use the encrypted fields
    handle(msg.value())
```

A tampered field produces `CORRUPT_MESSAGE` (2) on the partition, never a value.