# Kafka clients reference

> The decision per Kafka API, refusals at load, physical names, sessions, listener bounds, administration and troubleshooting.

Reference - Since version 1.6.0 - Flexway v1.7.1 - https://flexway.io/en/docs/operations-kafka-reference/

> Operations guide: configure, deploy and observe the gateway. What the feature does, product side: [product guide](https://flexway.io/en/docs/product-overview/); calling it: [developer guide](https://flexway.io/en/docs/developer-overview/).

The exact facts of the Kafka listener. What it does and why: [Native Kafka clients](https://flexway.io/en/docs/operations-kafka-clients/); declaring it: [Declare a Kafka virtual cluster](https://flexway.io/en/docs/operations-kafka-virtual-clusters/).

## The decision, by Kafka API

A closed table - an API missing from it is refused.

| API | Decision |
| --- | --- |
| `ApiVersions`, `SaslHandshake`, `SaslAuthenticate` | Answered by the gateway; `ApiVersions` announces only what it judges, at the intersection with the brokers' versions - never a version that names topics by id (`Produce` <= 12, `Fetch` <= 12, `Metadata` 1 to 12, `OffsetCommit` and `OffsetFetch` <= 9) |
| `Metadata`, `DescribeCluster` | The session's topics only, under their short names - a topic unknown or not opened is `UNKNOWN_TOPIC_OR_PARTITION`, never asked of the broker; brokers renamed (SNI or fallback port); cluster id = the virtual cluster's; automatic creation off; authorized operations never asked |
| `Produce` | A `send` operation; without one `TOPIC_AUTHORIZATION_FAILED`; record batches relayed **without being decoded** |
| `Fetch`, `ListOffsets`, `OffsetForLeaderEpoch` | A `receive` operation; `Fetch` bounded to `max_response_bytes / 2` and `max_fetch_wait`, always full (no incremental session); a replica's request is relayed as a consumer's (`ReplicaID` `-1`) |
| `FindCoordinator`, `JoinGroup`, `SyncGroup`, `Heartbeat`, `LeaveGroup`, `OffsetCommit`, `OffsetFetch`, `DescribeGroups`, `ListGroups` | A `receive` operation opens groups; `GROUP_AUTHORIZATION_FAILED` otherwise; `ListGroups` lists the tenant's only |
| `InitProducerId`, `AddPartitionsToTxn`, `AddOffsetsToTxn`, `EndTxn`, `TxnOffsetCommit` | Idempotent producer: a `send` operation; transactional: a `send` operation declaring `transactions: true` (`TRANSACTIONAL_ID_AUTHORIZATION_FAILED` otherwise) |
| `CreateTopics`, `DeleteTopics`, `DeleteRecords`, `CreatePartitions`, `DescribeConfigs`, `IncrementalAlterConfigs` (topics), `DeleteGroups`, `OffsetDelete` | **Refused** (`TOPIC_AUTHORIZATION_FAILED`, `GROUP_AUTHORIZATION_FAILED`) unless a token carries the operation's `admin_scope`; never a topic the contract does not declare; never a broker resource nor a group outside the tenant |
| ACLs, quotas, SCRAM credentials, reassignments, delegation tokens, `AlterConfigs`, broker-to-broker APIs, KIP-848 and KIP-932 groups, telemetry | **Never announced**; a client that sends one anyway is disconnected |

## Refused at load

Reported all at once: `x-flexway-kafka-proxy` without `listeners.kafka` (or the reverse), a
cluster name that is not a DNS label, a served operation without `security`, with a scheme the listener
does not verify (`scramSha256`, `gssapi`, `apiKey` outside the password), public, or with
`x-flexway-validation: full` (batches cross undecoded), `transactions` on a `receive` operation, a
channel whose address is not `<prefix>{tenant}.<name>` or whose key is not a topic name (nor `__`), two
prefixes on one upstream or on two upstreams reaching one broker, two channels whose names Kafka confuses (`.` read as `_`), a cluster on two upstreams, a token admitted without `auth.issuers`, a key
without `auth.api_keys`, a certificate without `client_ca_file`, fallback ports of a cluster the
contract does not declare, a served AsyncAPI operation's id equal to an OpenAPI operation's (a key bound
to one would open the other).

## Names of groups and transactions

A tenant's topics are written `<prefix><tenant>.<name>`, its consumer groups and transactional ids in
**sub-namespaces of their own**: `<prefix><tenant>.~g.<name>` and `<prefix><tenant>.~t.<name>`.
A Kafka topic name never holds `~`: no group of an application that names its groups after its topics, no
`flexway.ws.*` or `flexway.deliver.*` group of the gateway falls inside - the namespace is the listener's
alone. A served tenant is written in letters, digits and `-`: no dot (it ends the tenant in a name), no
underscore (Kafka reads `.` and `_` as one character in its metrics and refuses a topic that collides that
way).

| Client `acme` writes | The broker holds |
| --- | --- |
| topic `orders` | `acme.orders` (empty prefix) |
| `group.id=billing` | `acme.~g.billing` |
| `transactional.id=tx-orders` | `acme.~t.tx-orders` |
| `client.id=app-1` | `acme.app-1` (the broker's logs and quotas) |
| `ListGroups` | the `acme.~g.*` groups only, answered `billing`... |

### Broker ACLs

For the gateway's principal, empty prefix: `Topic` prefixed `acme.` (per tenant, or `*` when
the gateway is alone), `Group` prefixed `acme.~g.`, `TransactionalId` prefixed `acme.~t.`, `IdempotentWrite`.
Two upstreams reaching one broker under two prefixes are refused at load; addresses do not tell everything
(an alias, a load balancer, another listener of the same cluster), so at its first connection the listener
also reads the **cluster id** of each upstream's `Metadata`: a second upstream naming the same physical
cluster under another prefix is refused (log `kafka connection refused: two upstreams reach the same
physical cluster`, level Error). A broker that gives no cluster id (`Metadata` v0, v1) is told apart by its
addresses only.

## Sessions: revocation, lifetime, certificates

| Behaviour | Why | Default |
| --- | --- | --- |
| **Revocation on reload** | A key removed from the store, bound to other operations or to another tenant, a certificate removed from `auth.mtls.clients` or rebound, a token the issuers in force no longer accept: at the next request the open connection is **closed** - the client reconnects, and its new authentication is judged on the configuration in force | Always |
| `max_session_lifetime` | Every session is bounded, whatever its credential: a SASL client re-authenticates before it (`session_lifetime_ms` announced, KIP-368 - franz-go, librdkafka and Java do it on their own), a certificate's connection is closed at it; a revocation is seen at the latest then | `1h` (`1m` to `24h`) |
| A certificate's session | Bounded by the certificate's end of validity (`NotAfter`) too | Always |
| A token's session | **Verified again on every reload** by the issuers in force (the token is kept in memory for it, never written): an issuer removed from `auth.issuers`, an audience or a delegation bound changed, another principal behind it close the connection; bounded by its `exp` too | Always |

```yaml
listeners:
  kafka:
    max_session_lifetime: 30m      # a removed key, a revoked certificate: seen within 30 min at the latest
```

Remove the key `acme-orders-producer` from the store (the mounted `Secret`, read again on reload): the
producers opened with it see their connection closed at their next request, reconnect and fail
authentication (`SASL_AUTHENTICATION_FAILED`) - what to look for in their logs.

## Listener bounds

Each bound protects the gateway from a client - authenticated or not - that would take too much:

| Key | Why | Default |
| --- | --- | --- |
| `auth_timeout` | An **absolute** time to authenticate, TLS handshake included: a client that sends frames without ever authenticating is closed at it | `10s` |
| `max_unauthenticated_connections` | Places for connections not authenticated yet: past it a new one is closed at once - an anonymous flood never takes the authenticated clients' places | `256` |
| `max_connections_per_ip` | Connections of one address - a client opens one per broker, so a few dozen at most; below `max_unauthenticated_connections` so that one address cannot take every unauthenticated place. `0` behind a proxy that hides the clients' addresses (a `LoadBalancer` without `externalTrafficPolicy: Local`) | `16` |
| `max_connections_per_tenant` | Connections of one tenant, at authentication (`SASL_AUTHENTICATION_FAILED` past it): one tenant does not exhaust the listener for the others | `1024` |
| `max_connection_rate_per_tenant` | New connections of one tenant per second, at authentication (`SASL_AUTHENTICATION_FAILED` past it; a re-authentication takes none): a client reconnecting in a loop spends its tenant's handshakes, never the replica's. `0`: not bounded | `100` (provisional: to be set by performance measurements) |
| `max_fetch_wait` | How long a `Fetch` waits at the broker (the client's `max_wait_ms`, brought down to it): an upstream connection is never held without end | `30s` |
| `max_in_flight_bytes` | Broker answers held in memory per connection, read and not yet written: past it the broker is not read until the client reads (one answer always crosses alone) | `64MiB` |
| `max_tenant_in_flight_bytes` | The same bound for all the connections of **one tenant** - its share of the replica's: a tenant whose clients stop reading waits on its own share (or is closed, `on_tenant_budget_exceeded: close`), never on the replica's budget, so it cannot stall the other tenants (16 connections at `64MiB` would otherwise fill `1GiB`). From `max_in_flight_bytes` to `max_total_in_flight_bytes` (not reloadable) | an eighth of `max_total_in_flight_bytes` (`128MiB`) |
| `max_total_in_flight_bytes` | The same bound for the whole replica (not reloadable) | `1GiB` |
| Writing to the client | A client that does not read its answers for 30 s is closed, never waited for | `30s`, fixed |
| `max_topic_partitions` | Partitions an administrator may give a topic (`CreateTopics`, `CreatePartitions`) | `64` |

```yaml
listeners:
  kafka:
    address: ":9093"
    tls: { cert_file: /etc/flexway-kafka-tls/tls.crt, key_file: /etc/flexway-kafka-tls/tls.key }
    sni_domain: kafka.example.com
    auth_timeout: 10s
    max_connections: 4096
    max_unauthenticated_connections: 256
    max_connections_per_ip: 0            # behind a LoadBalancer that hides the addresses
    max_connections_per_tenant: 512
    max_connection_rate_per_tenant: 50   # new connections per second and tenant
    max_fetch_wait: 30s
    max_in_flight_bytes: 64MiB
    max_tenant_in_flight_bytes: 128MiB   # filling the replica takes eight tenants' clients, never one
    max_total_in_flight_bytes: 1GiB
    max_session_lifetime: 1h
    max_topic_partitions: 32
```

### Past a tenant's share

`on_tenant_budget_exceeded` (reloadable): what happens to a connection whose
next broker answer would take its tenant past `max_tenant_in_flight_bytes`. Either way the other tenants
are not stalled.

| Value | Behaviour | When |
| --- | --- | --- |
| `wait` (default) | The broker is not read for that connection until the tenant's clients read their answers - Kafka's own backpressure; bounded by the 30 s write timeout | Most tenants: a healthy consumer's burst (a catch-up, a large `Fetch`) is slowed, never cut |
| `close` | The connection is closed at once (log `kafka connection closed: the tenant's in-flight share is reached`, level Warn, with the tenant and the bound); what it held is given back; the client reconnects | Untrusted or hostile tenants, strict isolation: no tenant keeps the listener's memory waiting on clients that do not read |

```yaml
listeners:
  kafka:
    max_tenant_in_flight_bytes: 128MiB
    on_tenant_budget_exceeded: wait      # the default: bursts slowed, never cut
```

```yaml
listeners:
  kafka:
    max_tenant_in_flight_bytes: 64MiB
    on_tenant_budget_exceeded: close     # tenants you do not trust to read their answers
```

## Bounded administration

An operation's `admin_scope` opens the administration of **its** topic within fixed limits: `CreateTopics`
and `CreatePartitions` without replica assignment (placement is the cluster's) and with at most
`max_topic_partitions` partitions; `CreateTopics` and `IncrementalAlterConfigs` set only configurations of
the allowed list - `cleanup.policy`, `compression.type`, `delete.retention.ms`, `max.compaction.lag.ms`,
`min.compaction.lag.ms`, `min.cleanable.dirty.ratio`, `max.message.bytes`, `message.timestamp.type`,
`retention.bytes`, `retention.ms`, `segment.bytes`, `segment.ms`. Anything else
(`unclean.leader.election.enable`, `min.insync.replicas`, throttles, placement) answers `POLICY_VIOLATION`.
`DescribeConfigs` returns neither synonyms nor documentation (the broker's own settings).

Each value is bounded too, and the bounds are fixed - a topic's owner tunes within them, the cluster's
operator alone goes beyond (on the broker, not through the gateway):

| Configuration | Allowed values | Why |
| --- | --- | --- |
| `retention.ms` | `60000` (1 min) to `31536000000` (365 d); `-1` refused | Unlimited retention fills the brokers' disks: the operator's decision |
| `retention.bytes` | `1MiB` to `1TiB` (in bytes); `-1` refused | Same |
| `segment.ms` | `600000` (10 min) to 30 d | A segment that rolls every millisecond opens files by the thousand |
| `segment.bytes` | `1MiB` to `1GiB` | Same, by size |
| `max.compaction.lag.ms` | 1 h to 365 d | Compaction that never rests takes the broker's CPU |
| `min.compaction.lag.ms`, `delete.retention.ms` | `0` to 30 d | |
| `min.cleanable.dirty.ratio` | `0.1` to `1` | Same as `max.compaction.lag.ms` |
| `max.message.bytes` | `1KiB` to `max_request_bytes` | A larger message would not cross the listener anyway |
| `cleanup.policy` | `delete`, `compact` or both | |
| `compression.type` | `producer`, `uncompressed`, `gzip`, `snappy`, `lz4`, `zstd` | |
| `message.timestamp.type` | `CreateTime`, `LogAppendTime` | |

A `DELETE` (`IncrementalAlterConfigs`) back to the broker's default is always allowed. A value outside its
bounds answers `POLICY_VIOLATION`:

```bash
kafka-configs --bootstrap-server bootstrap.shop.kafka.example.com:9093 --command-config admin.properties \
  --alter --entity-type topics --entity-name orders --add-config retention.ms=-1
# -> refused, POLICY_VIOLATION; --add-config retention.ms=604800000 (7 days) is accepted
```

## Troubleshooting

Symptoms seen at the client, their cause and what to do.

| Symptom | Cause | Action |
| --- | --- | --- |
| `SASL_AUTHENTICATION_FAILED` | Unknown or expired key, PLAIN user other than the key's id, token invalid or without `exp`, tenant missing or holding `.`/`_`, a cluster where no operation admits the identity, `max_connections_per_tenant` or `max_connection_rate_per_tenant` reached | Log `kafka authentication refused` (the cause, never the key); `user = <key id>`, a token with `exp` and the tenant claim |
| `UNSUPPORTED_SASL_MECHANISM` | A mechanism not served (SCRAM, GSSAPI), or PLAIN/OAUTHBEARER without an alternative in the contract | A `plain` or `oauth2` alternative on the operation, `auth.api_keys` or `auth.issuers` |
| Connection closed right after the TCP connection | `max_unauthenticated_connections` or `max_connections_per_ip` reached, SNI naming no cluster, undeclared cluster | Log `kafka connection refused` (ten lines per second at most before authentication); `max_connections_per_ip: 0` behind a proxy |
| Connection closed after 10 s without authentication | `auth_timeout` | The client must authenticate within it |
| `UNKNOWN_TOPIC_OR_PARTITION` | Topic not declared in the contract, or opened to the identity by no operation | The channel in the contract, the operation bound to the key or the token's scope |
| `TOPIC_AUTHORIZATION_FAILED` | The topic exists but the direction is missing (producing with a `receive` operation only), administration without `admin_scope` | The `send` operation, or a token carrying the `admin_scope` |
| `CLUSTER_AUTHORIZATION_FAILED` on `InitProducerId` | Idempotent producer of a session that writes nothing | A `send` operation, or `enable.idempotence=false` for a client that does not produce |
| `TRANSACTIONAL_ID_AUTHORIZATION_FAILED` | No `send` operation declares `transactions: true` | `x-flexway-kafka-proxy: { transactions: true }` |
| `GROUP_AUTHORIZATION_FAILED` | Joining a group without a `receive` operation | An admitted `receive` operation |
| `POLICY_VIOLATION` | A configuration outside the list or a value outside its bounds (`retention.ms=-1`), a replica assignment, too many partitions; on `Produce`, the tenant's daily or monthly budget spent (log `kafka produce refused: the tenant's quota is spent`, with the window) | [Bounded administration](#bounded-administration); for the budget, the next window (UTC) or a raised `quota` policy ([Quotas: KIP-219 throttling](https://flexway.io/en/docs/operations-kafka-quotas/#quotas-kip-219-throttling)) |
| A producer or consumer slowed, without error | `throttle_time_ms`: a quota of `listeners.kafka.quotas`, or the broker's on the gateway's principal | `flexway_kafka_throttle_seconds` by `quota` and `scope` ([Signals: metrics, traces, audit](https://flexway.io/en/docs/operations-kafka-quotas/#signals-metrics-traces-audit)); raise the rate or `window`, or the broker's quota (`scope="broker"`) |
| Requests timing out at the client (`request.timeout.ms`) while throttled | `max_throttle` above the client's timeout | `max_throttle` below the clients' `request.timeout.ms` |
| Connection refused at once, log `two upstreams reach the same physical cluster` | Two upstreams with two prefixes on one physical cluster (same cluster id) | One upstream, one prefix per physical cluster |
| Connection closed along the way, log `the tenant's in-flight share is reached` (Warn) | `on_tenant_budget_exceeded: close` and the tenant's clients hold more unread answers than `max_tenant_in_flight_bytes` | Clients that read their answers, a larger share, or `wait` for trusted tenants |
| Connection closed along the way | Key, certificate or token revoked on reload, session expired without re-authentication, client not reading its answers for 30 s, API or version not advertised | The client reconnects; check its re-authentication (KIP-368) |
| The client loops on `broker-<id>... unmapped.invalid` | The fallback range does not cover that node id | Widen `ports` or adjust `first_node_id` |
| `KAFKA_STORAGE_ERROR`, `RESOURCE_NOT_FOUND` or `CORRUPT_MESSAGE` on an encrypted topic | KMS unreachable, key destroyed, or records naming a key that neither `kms.key` nor `kms.previous_keys` names | [Field encryption](https://flexway.io/en/docs/operations-kafka-encryption/), troubleshooting table |