Référence des clients Kafka

Guide d'exploitation : configurer, déployer et observer la passerelle. Ce que fait la fonctionnalité côté produit : guide produit ; l'appeler : guide développeur.

Les faits exacts de l'écouteur Kafka. Ce qu'il fait et pourquoi : Clients Kafka natifs ; le déclarer : Déclarer un cluster virtuel Kafka.

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

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

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

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

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 :

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 ; pour le budget, la fenêtre suivante (UTC) ou une politique quota relevée (Quotas : l'étranglement 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) ; 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, table de dépannage