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 field encryption; operations side: Native Kafka clients: a virtual cluster per tenant, Field 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

# 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

# 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}";

// 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:

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.