# Référence des clients Kafka

> La décision par API Kafka, les refus au chargement, les noms physiques, les sessions, les bornes de l'écouteur, l'administration et le dépannage.

Référence - Depuis la version 1.6.0 - Flexway v1.7.1 - https://flexway.io/docs/operations-kafka-reference/

> Guide d'exploitation : configurer, déployer et observer la passerelle. Ce que fait la fonctionnalité côté produit : [guide produit](https://flexway.io/docs/product-overview/) ; l'appeler : [guide développeur](https://flexway.io/docs/developer-overview/).

Les faits exacts de l'écouteur Kafka. Ce qu'il fait et pourquoi : [Clients Kafka natifs](https://flexway.io/docs/operations-kafka-clients/) ; le déclarer : [Déclarer un cluster virtuel Kafka](https://flexway.io/docs/operations-kafka-virtual-clusters/).

## La décision, par API Kafka

Table fermée - une API absente est refusée.

| API | Décision |
| --- | --- |
| `ApiVersions`, `SaslHandshake`, `SaslAuthenticate` | Répondues par la passerelle ; `ApiVersions` n'annonce que ce qu'elle juge, à l'intersection des versions des courtiers - jamais une version qui nomme les sujets par identifiant (`Produce` <= 12, `Fetch` <= 12, `Metadata` 1 à 12, `OffsetCommit` et `OffsetFetch` <= 9) |
| `Metadata`, `DescribeCluster` | Les sujets de la session seulement, sous leur nom court - un sujet inconnu ou non ouvert est `UNKNOWN_TOPIC_OR_PARTITION`, jamais demandé au courtier ; courtiers renommés (SNI ou port de repli) ; identifiant du cluster = celui du cluster virtuel ; création automatique coupée ; opérations autorisées jamais demandées |
| `Produce` | Opération `send` ; sans elle `TOPIC_AUTHORIZATION_FAILED` ; lots d'enregistrements relayés **sans être décodés** |
| `Fetch`, `ListOffsets`, `OffsetForLeaderEpoch` | Opération `receive` ; `Fetch` borné à `max_response_bytes / 2` et `max_fetch_wait`, toujours complet (sans session incrémentale) ; une requête de réplique est relayée comme celle d'un consommateur (`ReplicaID` `-1`) |
| `FindCoordinator`, `JoinGroup`, `SyncGroup`, `Heartbeat`, `LeaveGroup`, `OffsetCommit`, `OffsetFetch`, `DescribeGroups`, `ListGroups` | Une opération `receive` ouvre les groupes ; `GROUP_AUTHORIZATION_FAILED` sinon ; `ListGroups` ne liste que ceux du tenant |
| `InitProducerId`, `AddPartitionsToTxn`, `AddOffsetsToTxn`, `EndTxn`, `TxnOffsetCommit` | Producteur idempotent : une opération `send` ; transactionnel : une opération `send` qui déclare `transactions: true` (`TRANSACTIONAL_ID_AUTHORIZATION_FAILED` sinon) |
| `CreateTopics`, `DeleteTopics`, `DeleteRecords`, `CreatePartitions`, `DescribeConfigs`, `IncrementalAlterConfigs` (sujets), `DeleteGroups`, `OffsetDelete` | **Refusées** (`TOPIC_AUTHORIZATION_FAILED`, `GROUP_AUTHORIZATION_FAILED`) sauf jeton portant l'`admin_scope` de l'opération ; jamais un sujet que le contrat ne déclare pas ; jamais une ressource du courtier ou d'un groupe hors du tenant |
| ACL, quotas, identifiants SCRAM, réaffectations, jetons de délégation, `AlterConfigs`, API des courtiers entre eux, groupes de KIP-848 et KIP-932, télémétrie | **Jamais annoncées** ; un client qui les envoie quand même est déconnecté |

## Refusé au chargement

Signalés tous d'un coup : `x-flexway-kafka-proxy` sans `listeners.kafka` (ou
l'inverse), un nom de cluster qui n'est pas une étiquette DNS, une opération servie sans `security`,
avec un schéma que l'écouteur ne vérifie pas (`scramSha256`, `gssapi`, `apiKey` hors du mot de
passe), publique, ou `x-flexway-validation: full` (les lots passent sans être décodés), `transactions`
sur une opération `receive`, un canal dont l'adresse n'est pas `<préfixe>{tenant}.<nom>` ou dont la clé
n'est pas un nom de sujet (ni `__`), deux canaux dont Kafka confondrait les noms (`.` lu comme `_`),
deux préfixes sur un même amont ou sur deux amonts qui joignent un même courtier, un cluster sur deux amonts, un jeton
admis sans `auth.issuers`, une clé sans `auth.api_keys`, un certificat sans `client_ca_file`, une plage de
repli d'un cluster que le contrat ne déclare pas, un identifiant d'opération AsyncAPI servie égal à
celui d'une opération OpenAPI (une clé liée à l'un ouvrirait l'autre).

## Les noms des groupes et des transactions

Les sujets d'un tenant sont écrits `<préfixe><tenant>.<nom>`, ses groupes de consommateurs et ses
identifiants transactionnels dans des **sous-espaces à eux** : `<préfixe><tenant>.~g.<nom>`
et `<préfixe><tenant>.~t.<nom>`. Un nom de sujet Kafka ne contient jamais `~` : aucun groupe d'une
application qui nomme ses groupes comme ses sujets, aucun groupe `flexway.ws.*` ou `flexway.deliver.*` de la
passerelle n'y tombe - l'espace est celui de l'écouteur seul. Un tenant servi s'écrit en lettres, chiffres
et `-` : ni point (il termine le tenant dans un nom), ni souligné (Kafka lit `.` et `_` comme un même
caractère dans ses métriques et refuse un sujet qui en heurte un autre ainsi).

| Le client `acme` écrit | Le courtier tient |
| --- | --- |
| sujet `orders` | `acme.orders` (préfixe vide) |
| `group.id=billing` | `acme.~g.billing` |
| `transactional.id=tx-orders` | `acme.~t.tx-orders` |
| `client.id=app-1` | `acme.app-1` (journaux et quotas du courtier) |
| `ListGroups` | les seuls groupes `acme.~g.*`, rendus `billing`... |

### ACL du courtier

Pour le principal de la passerelle, préfixe vide : `Topic` préfixe `acme.` (pour chaque
tenant, ou `*` si la passerelle est seule), `Group` préfixe `acme.~g.`, `TransactionalId` préfixe `acme.~t.`,
`IdempotentWrite`. Deux amonts qui joignent un même courtier sous deux préfixes sont refusés au chargement ;
les adresses ne disent pas tout (un alias, un répartiteur, un autre écouteur du même cluster), alors à sa
première connexion l'écouteur lit aussi l'**identifiant du cluster** dans le `Metadata` de chaque amont : un
second amont qui nomme le même cluster physique sous un autre préfixe est refusé (journal `kafka connection
refused: two upstreams reach the same physical cluster`, niveau Error). Un courtier qui ne donne pas
d'identifiant (`Metadata` v0, v1) n'est distingué que par ses adresses.

## Sessions : révocation, durée, certificats

| Comportement | Pourquoi | Défaut |
| --- | --- | --- |
| **Révocation au rechargement** | Une clé retirée du magasin, rattachée à d'autres opérations ou à un autre tenant, un certificat retiré de `auth.mtls.clients` ou rebranché, un jeton que les émetteurs en vigueur n'acceptent plus : à la requête suivante, la connexion ouverte est **fermée** - le client se reconnecte, et sa nouvelle authentification est jugée sur la configuration en vigueur | Toujours |
| `max_session_lifetime` | Toute session est bornée, quel que soit son justificatif : un client SASL se ré-authentifie avant (`session_lifetime_ms` annoncé, KIP-368 - franz-go, librdkafka et Java le font seuls), une connexion par certificat est fermée à son terme ; une révocation est vue au plus tard alors | `1h` (de `1m` à `24h`) |
| Session d'un certificat | Bornée aussi par la fin de validité du certificat (`NotAfter`) | Toujours |
| Session d'un jeton | **Revérifiée à chaque rechargement** par les émetteurs en vigueur (le jeton est gardé en mémoire pour cela, jamais écrit) : un émetteur retiré d'`auth.issuers`, une audience ou une borne de délégation changée, un autre principal derrière lui ferment la connexion ; bornée aussi par son `exp` | Toujours |

```yaml
listeners:
  kafka:
    max_session_lifetime: 30m      # une clé retirée, un certificat révoqué : vus au plus tard en 30 min
```

Retirer la clé `acme-orders-producer` du magasin (le `Secret` monté, relu au rechargement) : les
producteurs ouverts avec elle reçoivent une connexion fermée à leur requête suivante, se reconnectent et
échouent à l'authentification (`SASL_AUTHENTICATION_FAILED`) - ce qu'il faut voir dans leurs journaux.

## Bornes de l'écouteur

Chaque borne protège la passerelle d'un client - authentifié ou non - qui en consommerait trop :

| Clé | Pourquoi | Défaut |
| --- | --- | --- |
| `auth_timeout` | Délai **absolu** pour s'authentifier, poignée TLS comprise : un client qui envoie des trames sans jamais s'authentifier est fermé à son terme | `10s` |
| `max_unauthenticated_connections` | Places des connexions pas encore authentifiées : au-delà, une nouvelle est fermée aussitôt - un afflux anonyme ne prend jamais la place des clients authentifiés | `256` |
| `max_connections_per_ip` | Connexions d'une même adresse - un client en ouvre une par courtier, quelques dizaines au plus ; sous `max_unauthenticated_connections` pour qu'une adresse ne prenne pas toutes les places non authentifiées. `0` derrière un proxy qui masque les adresses des clients (un `LoadBalancer` sans `externalTrafficPolicy: Local`) | `16` |
| `max_connections_per_tenant` | Connexions d'un même tenant, à l'authentification (`SASL_AUTHENTICATION_FAILED` au-delà) : un tenant n'épuise pas l'écouteur des autres | `1024` |
| `max_connection_rate_per_tenant` | Nouvelles connexions d'un même tenant par seconde, à l'authentification (`SASL_AUTHENTICATION_FAILED` au-delà ; une ré-authentification n'en prend aucune) : un client qui se reconnecte en boucle dépense les poignées de main de son tenant, jamais celles de la réplique. `0` : sans borne | `100` (provisoire : à fixer par les mesures de performance) |
| `max_fetch_wait` | Attente d'un `Fetch` chez le courtier (`max_wait_ms` du client, ramené à cette borne) : une connexion amont n'est jamais tenue indéfiniment | `30s` |
| `max_in_flight_bytes` | Réponses du courtier tenues en mémoire par connexion, lues et pas encore écrites : au-delà, le courtier n'est plus lu tant que le client ne lit pas (une réponse passe toujours seule) | `64MiB` |
| `max_tenant_in_flight_bytes` | La même borne pour toutes les connexions d'**un tenant** - sa part de celle de la réplique : un tenant dont les clients ne lisent plus attend sur sa propre part (ou est fermé, `on_tenant_budget_exceeded: close`), jamais sur le budget de la réplique, et ne bloque pas les autres tenants (16 connexions à `64MiB` rempliraient sinon `1GiB`). De `max_in_flight_bytes` à `max_total_in_flight_bytes` (non rechargeable) | un huitième de `max_total_in_flight_bytes` (`128MiB`) |
| `max_total_in_flight_bytes` | La même borne pour toute la réplique (non rechargeable) | `1GiB` |
| Écriture vers le client | Un client qui ne lit pas ses réponses pendant 30 s est fermé, jamais attendu | `30s`, fixe |
| `max_topic_partitions` | Partitions qu'un administrateur peut donner à un sujet (`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            # derrière un LoadBalancer qui masque les adresses
    max_connections_per_tenant: 512
    max_connection_rate_per_tenant: 50   # nouvelles connexions par seconde et par tenant
    max_fetch_wait: 30s
    max_in_flight_bytes: 64MiB
    max_tenant_in_flight_bytes: 128MiB   # remplir la réplique demande les clients de huit tenants, jamais d'un seul
    max_total_in_flight_bytes: 1GiB
    max_session_lifetime: 1h
    max_topic_partitions: 32
```

### Au-delà de la part d'un tenant

`on_tenant_budget_exceeded` (rechargeable) : ce que devient une
connexion dont la prochaine réponse du courtier ferait passer son tenant au-delà de
`max_tenant_in_flight_bytes`. Dans les deux cas, les autres tenants ne sont pas arrêtés.

| Valeur | Comportement | Quand |
| --- | --- | --- |
| `wait` (défaut) | Le courtier n'est plus lu pour cette connexion tant que les clients du tenant ne lisent pas leurs réponses - la contre-pression de Kafka lui-même ; borné par le délai d'écriture de 30 s | La plupart des tenants : la rafale d'un consommateur sain (un rattrapage, un gros `Fetch`) est ralentie, jamais coupée |
| `close` | La connexion est fermée aussitôt (journal `kafka connection closed: the tenant's in-flight share is reached`, niveau Warn, avec le tenant et la borne) ; ce qu'elle tenait est rendu ; le client se reconnecte | Tenants non fiables ou hostiles, isolation stricte : aucun tenant ne fait attendre la mémoire de l'écouteur sur des clients qui ne lisent pas |

```yaml
listeners:
  kafka:
    max_tenant_in_flight_bytes: 128MiB
    on_tenant_budget_exceeded: wait      # le défaut : rafales ralenties, jamais coupées
```

```yaml
listeners:
  kafka:
    max_tenant_in_flight_bytes: 64MiB
    on_tenant_budget_exceeded: close     # tenants dont on ne garantit pas qu'ils lisent leurs réponses
```

## Administration bornée

L'`admin_scope` d'une opération ouvre l'administration de **son** sujet, dans des limites fixes :
`CreateTopics` et `CreatePartitions` sans affectation de répliques (le placement est celui du cluster) et
au plus `max_topic_partitions` partitions ; `CreateTopics` et `IncrementalAlterConfigs` ne posent que des
configurations de la liste permise - `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`. Le reste
(`unclean.leader.election.enable`, `min.insync.replicas`, étranglements, placement) répond
`POLICY_VIOLATION`. `DescribeConfigs` ne rend ni synonymes ni documentation (les réglages du courtier).

Chaque valeur est bornée aussi, et les bornes sont fixes - le propriétaire d'un sujet règle à l'intérieur,
l'exploitant du cluster seul va au-delà (sur le courtier, pas par la passerelle) :

| Configuration | Valeurs permises | Pourquoi |
| --- | --- | --- |
| `retention.ms` | `60000` (1 min) à `31536000000` (365 j) ; `-1` refusé | Une rétention illimitée remplit les disques des courtiers : décision de l'exploitant |
| `retention.bytes` | `1MiB` à `1TiB` (en octets) ; `-1` refusé | Idem |
| `segment.ms` | `600000` (10 min) à 30 j | Un segment qui bascule chaque milliseconde ouvre des fichiers par milliers |
| `segment.bytes` | `1MiB` à `1GiB` | Idem, par taille |
| `max.compaction.lag.ms` | 1 h à 365 j | Une compaction qui ne se repose jamais prend le CPU du courtier |
| `min.compaction.lag.ms`, `delete.retention.ms` | `0` à 30 j | |
| `min.cleanable.dirty.ratio` | `0.1` à `1` | Comme `max.compaction.lag.ms` |
| `max.message.bytes` | `1KiB` à `max_request_bytes` | Un message plus gros ne traverserait pas l'écouteur de toute façon |
| `cleanup.policy` | `delete`, `compact` ou les deux | |
| `compression.type` | `producer`, `uncompressed`, `gzip`, `snappy`, `lz4`, `zstd` | |
| `message.timestamp.type` | `CreateTime`, `LogAppendTime` | |

Un `DELETE` (`IncrementalAlterConfigs`) qui revient à la valeur par défaut du courtier est toujours
permis. Une valeur hors de ses bornes répond `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
# -> refusé, POLICY_VIOLATION ; --add-config retention.ms=604800000 (7 jours) est accepté
```

## Dépannage

Symptômes vus côté client, leur cause et ce qu'il faut faire.

| Symptôme | Cause | Action |
| --- | --- | --- |
| `SASL_AUTHENTICATION_FAILED` | Clé inconnue ou expirée, utilisateur PLAIN autre que l'identifiant de la clé, jeton invalide ou sans `exp`, tenant absent ou avec `.`/`_`, cluster où aucune opération n'admet l'identité, `max_connections_per_tenant` ou `max_connection_rate_per_tenant` atteint | Journal `kafka authentication refused` (cause, jamais la clé) ; `user = <id de la clé>`, un jeton avec `exp` et la claim du tenant |
| `UNSUPPORTED_SASL_MECHANISM` | Un mécanisme non servi (SCRAM, GSSAPI), ou PLAIN/OAUTHBEARER sans alternative au contrat | Une alternative `plain` ou `oauth2` sur l'opération, `auth.api_keys` ou `auth.issuers` |
| Connexion fermée juste après la connexion TCP | `max_unauthenticated_connections` ou `max_connections_per_ip` atteint, SNI qui ne nomme aucun cluster, cluster non déclaré | Journal `kafka connection refused` (limité à 10 lignes par seconde avant l'authentification) ; `max_connections_per_ip: 0` derrière un proxy |
| Connexion fermée après 10 s sans authentification | `auth_timeout` | Le client doit s'authentifier dans le délai |
| `UNKNOWN_TOPIC_OR_PARTITION` | Sujet non déclaré au contrat, ou qu'aucune opération n'ouvre à l'identité | Le canal au contrat, l'opération liée à la clé ou la portée du jeton |
| `TOPIC_AUTHORIZATION_FAILED` | Le sujet existe mais le sens manque (produire avec une opération `receive` seule), administration sans `admin_scope` | L'opération `send`, ou un jeton portant l'`admin_scope` |
| `CLUSTER_AUTHORIZATION_FAILED` à `InitProducerId` | Producteur idempotent d'une session qui n'écrit rien | Une opération `send`, ou `enable.idempotence=false` pour un client qui ne produit pas |
| `TRANSACTIONAL_ID_AUTHORIZATION_FAILED` | Aucune opération `send` ne déclare `transactions: true` | `x-flexway-kafka-proxy: { transactions: true }` |
| `GROUP_AUTHORIZATION_FAILED` | Rejoindre un groupe sans opération `receive` | Une opération `receive` admise |
| `POLICY_VIOLATION` | Configuration hors liste ou valeur hors de ses bornes (`retention.ms=-1`), affectation de répliques, trop de partitions ; à `Produce`, le budget jour ou mois du tenant épuisé (journal `kafka produce refused: the tenant's quota is spent`, avec la fenêtre) | [Administration bornée](#administration-bornée) ; pour le budget, la fenêtre suivante (UTC) ou une politique `quota` relevée ([Quotas : l'étranglement de KIP-219](https://flexway.io/docs/operations-kafka-quotas/#quotas--l%C3%A9tranglement-de-kip-219)) |
| Producteur ou consommateur ralenti, sans erreur | `throttle_time_ms` : un quota de `listeners.kafka.quotas`, ou celui du courtier sur le principal de la passerelle | `flexway_kafka_throttle_seconds` par `quota` et `scope` ([Signaux : métriques, traces, audit](https://flexway.io/docs/operations-kafka-quotas/#signaux--m%C3%A9triques-traces-audit)) ; relever le débit ou `window`, ou le quota du courtier (`scope="broker"`) |
| Requêtes expirées côté client (`request.timeout.ms`) pendant un étranglement | `max_throttle` au-dessus du délai du client | `max_throttle` sous le `request.timeout.ms` des clients |
| Connexion refusée aussitôt, journal `two upstreams reach the same physical cluster` | Deux amonts à deux préfixes sur un même cluster physique (même identifiant) | Un amont, un préfixe par cluster physique |
| Connexion fermée en cours de route, journal `the tenant's in-flight share is reached` (Warn) | `on_tenant_budget_exceeded: close` et des clients du tenant qui tiennent plus de réponses non lues que `max_tenant_in_flight_bytes` | Des clients qui lisent leurs réponses, une part plus grande, ou `wait` pour des tenants de confiance |
| Connexion fermée en cours de route | Clé, certificat ou jeton révoqué au rechargement, session expirée sans ré-authentification, client qui ne lit pas ses réponses 30 s, API ou version non annoncée | Le client se reconnecte ; vérifier sa ré-authentification (KIP-368) |
| Le client tourne sur `broker-<id>... unmapped.invalid` | La plage de repli ne couvre pas cet identifiant de nœud | Élargir `ports` ou ajuster `first_node_id` |
| `KAFKA_STORAGE_ERROR`, `RESOURCE_NOT_FOUND` ou `CORRUPT_MESSAGE` sur un sujet chiffré | KMS injoignable, clé détruite, ou enregistrements qui nomment une clé que `kms.key` et `kms.previous_keys` ne nomment pas | [Chiffrement par champ](https://flexway.io/docs/operations-kafka-encryption/), table de dépannage |