Flexway - Clients Kafka natifs : clusters virtuels… (Exploitation)

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

Les producteurs, consommateurs et outils d'administration Kafka - n'importe quelle bibliothèque : franz-go, librdkafka (confluent-kafka-go, Python, .NET), le client Java - se branchent sur la passerelle au lieu du courtier, sans changer de code. Chaque tenant voit son cluster virtuel : les seuls sujets que le contrat déclare et que ses opérations lui ouvrent, sous leur nom court. Le client écrit dans orders, le courtier reçoit acme.orders ; groupes de consommateurs et identifiants transactionnels sont écrits dans des sous-espaces du tenant (acme.~g.billing, acme.~t.tx-orders, Les noms des groupes et des transactions), l'identifiant du client aussi (acme.app-1) - jamais un nom d'un autre tenant, ni dans une réponse ni vers le courtier. La passerelle se présente elle-même au courtier (TLS et SASL de l'amont) : le client ne connaît jamais ses identifiants. Disponible depuis 1.6.0.

Pas encore complet - les identités, les ACL et les quotas d'octets et de requêtes sont servis et testés de bout en bout. Deux points restent ouverts, sans version annoncée :

le budget jour ou mois d'un tenant ne compte que les écritures (Produce) : les lectures (Fetch) n'y sont pas comptées tant que leur code de refus n'est pas vérifié contre les principales bibliothèques clientes (Quotas) ;

les valeurs par défaut de max_throttle et de max_connection_rate_per_tenant sont provisoires, en attendant les mesures de performance : elles peuvent changer dans une version mineure - si vous en dépendez, écrivez-les dans votre configuration.

# Kafka clients behind the gateway (1.6.0): producers, consumers and admin tools of any
# library reach the virtual cluster "shop" the AsyncAPI contract declares - each tenant inside its own
# names, under the same identities as HTTP. Here the fallback inside a mesh, a port per broker; the TLS
# listener, brokers named by SNI, is in the guide.
version: 1
contracts:
  asyncapi: ./asyncapi.yaml
upstreams:
  events:
    protocol: kafka
    brokers: [kafka-0.kafka:9093, kafka-1.kafka:9093, kafka-2.kafka:9093]
    tls: { server_name: kafka.internal }
    sasl: { mechanism: SCRAM-SHA-512, username: flexway, password: "file:/etc/flexway-secrets/kafka/password" }
    timeout: 5s                                  # a connection to a broker, TLS and SASL included
auth:
  issuers:
    - name: customers
      discovery: https://login.example.net/.well-known/openid-configuration
      audience: kafka
      tenant_claim: org                          # every topic, group and transaction is the tenant's
  api_keys:
    store: "file:./api-keys.yaml"                # PLAIN: the key's id as the user, the key as the password
listeners:
  kafka:
    plaintext:
      mesh_only: true                            # clear text only where the mesh encrypts between pods
      advertised_host: flexway-kafka.flexway.svc.cluster.local
      clusters:
        shop: { ports: { from: 19092, to: 19095 }, first_node_id: 0 }   # bootstrap 19092, node n at 19093+n

# examples/kafka-proxy/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Shop events, version: "1.0.0" }
channels:
  orders:
    address: "{tenant}.orders"
    parameters:
      tenant: { x-flexway-bind: tenant }
    messages:
      order: { payload: { type: object, required: [id], properties: { id: { type: string } } } }
operations:
  produceOrders:
    action: send
    channel: { $ref: "#/channels/orders" }
    x-flexway-upstream: events
    x-flexway-kafka-proxy: { cluster: shop, transactions: true }
    security:
      - $ref: "#/components/securitySchemes/apiKey"
      - $ref: "#/components/securitySchemes/ordersWrite"
  consumeOrders:
    action: receive
    channel: { $ref: "#/channels/orders" }
    x-flexway-upstream: events
    x-flexway-kafka-proxy: { cluster: shop, admin_scope: "orders:admin" }
    security:
      - $ref: "#/components/securitySchemes/apiKey"
      - $ref: "#/components/securitySchemes/ordersRead"
components:
  securitySchemes:
    apiKey: { type: plain }                      # SASL PLAIN: an API key bound to the operation
    ordersWrite:                                 # SASL OAUTHBEARER: a token carrying the scope
      type: oauth2
      flows: { clientCredentials: { tokenUrl: "https://login.example.net/token", availableScopes: { "orders:write": Produce orders } } }
      scopes: ["orders:write"]
    ordersRead:
      type: oauth2
      flows: { clientCredentials: { tokenUrl: "https://login.example.net/token", availableScopes: { "orders:read": Read orders } } }
      scopes: ["orders:read"]

La convention du contrat (commune aux écouteurs : celle de x-flexway-device au jalon 1.7.0) :

Où	Forme	Sens
Opération AsyncAPI	`x-flexway-kafka-proxy: { cluster, transactions, admin_scope }`	L'opération est servie aux clients Kafka du cluster virtuel `cluster` (une étiquette DNS) ; `transactions: true` (opération `send`) ouvre le producteur transactionnel ; `admin_scope` ouvre l'administration du sujet à un jeton qui porte cette portée
`action`	`send` : le client produit (`Produce`) ; `receive` : il lit (`Fetch`, positions, groupes)	Le même sens qu'au plan asynchrone : vu du courtier, `send` écrit dans le canal
Canal	Sa clé est le nom du sujet que voit le client ; son `address` est `<préfixe>{tenant}.<nom>`, son seul paramètre lié au tenant	Le préfixe (vide, ou finissant par un point) est le même pour tous les canaux servis d'un amont ; groupes et identifiants transactionnels sont écrits `<préfixe><tenant>.<nom>`
`security` de l'opération	Schémas AsyncAPI 3 : `oauth2` ou `openIdConnect` avec `scopes`, `plain` (ou `apiKey` avec `in: password`), `X509`	Une alternative suffit ; lue pour les opérations servies aux clients Kafka seulement - une porte HTTP garde la sécurité de son opération OpenAPI

Les identités - les mêmes qu'en HTTP :

Mécanisme du client	Alternative au contrat	Ce que la passerelle vérifie
SASL `OAUTHBEARER`	`oauth2` / `openIdConnect` + `scopes`	Le jeton comme un `Bearer` : émetteur déclaré, audience, signature, chaîne de délégation ; `exp` obligatoire : la session dure jusqu'à lui, annoncée au client (`session_lifetime_ms`, KIP-368) - il se ré-authentifie avant, sinon la connexion est fermée ; une ré-authentification prouve le même principal
SASL `PLAIN`	`plain` (ou `apiKey`, `in: password`)	La clé d'API par son empreinte (Clés d'API) : le nom d'utilisateur est l'identifiant de la clé, le mot de passe la clé ; liée à son tenant et à ses opérations, bornée par son `expires_at`
Certificat client (mTLS)	`X509`	Un certificat vérifié par l'écouteur TLS (`listeners.kafka.tls.client_ca_file`) et déclaré dans `auth.mtls.clients` (Certificats clients (mTLS)), lié à son tenant et à ses opérations ; aucune poignée SASL

Une identité sans tenant, ou dont le tenant ne s'écrit pas en lettres, chiffres et - (un point, un souligné), est refusée à l'authentification - Les noms des groupes et des transactions.

L'écouteur TLS, les courtiers nommés par SNI : un seul port, le nom SNI dit quel courtier est visé - bootstrap.<cluster>.<sni_domain> pour l'amorçage, broker-<id>.<cluster>.<sni_domain> pour chacun ; Metadata et FindCoordinator répondent ces noms. Le certificat couvre *.<cluster>.<sni_domain> pour chaque cluster virtuel (un joker ne couvre qu'une étiquette) ; tous ces noms résolvent vers la passerelle (un enregistrement DNS joker, ou le Service LoadBalancer du chart).

# flexway.yaml - l'écouteur TLS ; le reste comme l'exemple ci-dessus
listeners:
  kafka:
    address: ":9093"
    tls:
      cert_file: /etc/flexway-kafka-tls/tls.crt        # *.shop.kafka.example.com
      key_file: /etc/flexway-kafka-tls/tls.key
      client_ca_file: /etc/flexway-kafka-tls/ca.crt    # facultatif : les certificats clients (X509)
    sni_domain: kafka.example.com
    advertised_port: 9093                # le port que voit le client (celui du LoadBalancer)
    max_request_bytes: 16MiB             # une requête authentifiée ; 64 Kio avant l'authentification
    max_response_bytes: 64MiB            # une réponse du courtier ; un Fetch est tenu sous la moitié
    max_connections: 4096
    auth_timeout: 10s
    idle_timeout: 10m
auth:
  mtls:
    clients:
      - { id: billing, subject: "CN=billing-app", tenant: acme, operations: [consumeOrders] }

Le repli en clair, un port par courtier (l'exemple) : sans TLS, SNI n'existe pas ; chaque cluster virtuel reçoit une plage - from pour l'amorçage, from + 1 + n pour le nœud first_node_id + n -, annoncée sous advertised_host. Refusé au démarrage sans mesh_only: true : le clair ne traverse qu'un maillage qui chiffre (Linkerd, Istio en mTLS strict). La plage couvre chaque identifiant de nœud du cluster, sinon le client reçoit une adresse injoignable et le journal le dit.

Les amonts - Apache Kafka (apache/kafka, Confluent Platform, MSK, Strimzi) et Redpanda se déclarent pareil ; la passerelle lit les versions de chaque courtier et n'annonce au client que l'intersection avec ce qu'elle juge :

upstreams:
  events:                                # Apache Kafka : SCRAM sur TLS
    protocol: kafka
    brokers: [kafka-0.kafka:9093, kafka-1.kafka:9093, kafka-2.kafka:9093]
    tls: { ca_file: /etc/flexway-secrets/kafka/ca.crt, server_name: kafka.internal }
    sasl: { mechanism: SCRAM-SHA-512, username: flexway, password: "file:/etc/flexway-secrets/kafka/password" }
  events-redpanda:                       # Redpanda : SASL SCRAM-SHA-256, ou mTLS de la passerelle
    protocol: kafka
    brokers: [redpanda-0.redpanda:9093]
    tls: { ca_file: /etc/flexway-secrets/redpanda/ca.crt, cert_file: /etc/flexway-secrets/redpanda/tls.crt, key_file: /etc/flexway-secrets/redpanda/tls.key }
    timeout: 5s                          # une connexion au courtier, TLS et SASL compris

Les clients - des propriétés Kafka standard, pointées sur la passerelle :

// franz-go : une clé d'API en SASL PLAIN sur TLS
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()),
	kgo.TransactionalID("tx-orders"), // devient acme.~t.tx-orders chez le courtier
)
// un jeton : kgo.SASL(oauth.Oauth(func(ctx context.Context) (oauth.Auth, error) { return oauth.Auth{Token: tok}, nil }))

# librdkafka (confluent-kafka-go, Python, .NET) : un jeton OAUTHBEARER de l'émetteur déclaré
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                         # devient acme.~g.billing

# Client Java (producteur, consommateur, AdminClient, Kafka Connect) : une clé d'API en 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}";
# un certificat client à la place : security.protocol=SSL, ssl.keystore.location=..., ssl.truststore.location=...
# group.protocol=classic (par défaut) : le protocole de groupe de KIP-848 n'est pas servi

La décision, par API Kafka - table fermée : une API absente est refusée (domain/kafkaproxy, tenue égale à la liste de franz-go par un test) :

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, tout 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).

Rechargement : l'adresse, TLS, sni_domain, advertised_port, la plage de repli et max_connections, max_total_in_flight_bytes et max_tenant_in_flight_bytes sont lus au démarrage (listeners.kafka refusé au rechargement s'ils changent) ; le contrat, les identités, les bornes et les délais s'appliquent à la requête suivante - une connexion garde son identité tant que son justificatif reste celui de la configuration en vigueur (Sessions : révocation, durée, certificats), et chaque requête est jugée sur celle-ci.

Une connexion amont par connexion cliente et par courtier : la session Kafka (producteur idempotent, transactions, étranglement du courtier) n'est jamais partagée entre clients ; l'arrêt draine (les requêtes en vol répondues, puis la connexion fermée : le client se reconnecte à une autre réplique). ACL minimales du courtier pour le principal de la passerelle : lire, écrire et décrire les sujets <préfixe><tenant>.* servis, les groupes <préfixe><tenant>.~g.* et identifiants transactionnels <préfixe><tenant>.~t.*, IdempotentWrite ; l'administration seulement si un admin_scope est déclaré. La NetworkPolicy des courtiers n'admet que la passerelle : aucun client ne la contourne avec les adresses réelles, que Metadata ne donne jamais.

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 (décision D-K7) : <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é

Ce qui ne traverse jamais

Vers le client	Vers le courtier
Les messages d'erreur du courtier (`ErrorMessage`, erreurs par enregistrement) : il peut y nommer un sujet physique, un groupe, une adresse - le code passe, les mots non	Les champs étiquetés inconnus de la requête du client
L'identifiant du cluster physique : `Metadata` et `DescribeCluster` rendent le nom du cluster virtuel	Une requête de réplique (`ReplicaID` forcé à `-1`)
L'adresse du membre (`ClientHost` de `DescribeGroups`)	Une session de `Fetch` incrémentale : chaque `Fetch` est complet, jugé seul
Les champs étiquetés inconnus de la réponse

Le chart : clair dans le maillage seulement

Les ports en clair vont dans un Service à part, <release>-flexway-kafka-mesh, toujours ClusterIP ; le rendu échoue sans kafka.networkPolicy.from (les pods ou namespaces du maillage).

# rendu : le port TLS sur un LoadBalancer, la plage en clair réservée au maillage
config:
  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
      plaintext: { mesh_only: true, advertised_host: orders-flexway-kafka-mesh.apps.svc.cluster.local, clusters: { shop: { ports: { from: 19092, to: 19095 } } } }
kafka:
  service: { type: LoadBalancer }
  tls: { clusters: [shop], issuerRef: { name: kafka-ca, kind: ClusterIssuer } }
  networkPolicy:
    from: [ { namespaceSelector: { matchLabels: { kubernetes.io/metadata.name: apps } } } ]

# refusé : du clair sans pairs déclarés
kafka:
  networkPolicy: { from: [] }
# Error: execution error at (flexway/templates/kafka.yaml): listeners.kafka.plaintext opens ports in
# clear: kafka.networkPolicy.enabled and kafka.networkPolicy.from (the mesh's pods or namespaces) are
# required - clear text is for the mesh only

Quotas : l'étranglement de KIP-219

Octets produits, octets lus et requêtes par seconde, pour chaque principal vérifié et pour chaque tenant (listeners.kafka.quotas). Jamais une coupure : au-delà, la réponse part tout de suite avec throttle_time_ms, puis la passerelle ne lit plus la connexion du client pendant ce délai (KIP-219). Un client qui ignore le champ - librdkafka ne l'a jamais appliqué (librdkafka#5592) - est freiné quand même : ce qu'il envoie attend dans la socket. Sans quotas, rien n'est étranglé.

Clé	Pourquoi	Défaut
`client.producer_byte_rate`	Octets produits par seconde (requêtes `Produce`, en-têtes compris) par un principal : le sujet du jeton, la clé d'API ou le certificat - jamais le `client.id`, qu'un client choisit librement	aucune borne
`client.consumer_byte_rate`	Octets lus par seconde (réponses `Fetch`) par un principal	aucune borne
`client.request_rate`	Requêtes par seconde d'un principal, toutes API, une fois authentifié	aucune borne
`tenant.producer_byte_rate`, `tenant.consumer_byte_rate`, `tenant.request_rate`	Les mêmes pour tous les principaux d'un tenant ensemble : un tenant n'achète pas plus de débit en multipliant les clés	aucune borne
`window`	La rafale admise d'un coup : un seau tient `débit x window` (les fenêtres de quota de Kafka font environ une seconde)	`1s` (de `100ms` à `10s`)
`max_throttle`	Plafond de `throttle_time_ms` - et du silence qui suit : sous le `request.timeout.ms` des clients (30 s par défaut chez Kafka), qu'un étranglement ne doit jamais faire expirer	`10s` (de `100ms` à `30s` ; défaut provisoire, à fixer par les mesures de performance)
`quota`	Le nom d'une politique `quota` (Poser des quotas) : le budget jour ou mois de requêtes `Produce` du tenant, compté sous ce nom comme en HTTP ; épuisé, le `Produce` est refusé `POLICY_VIOLATION` (44, non réessayable) et rien n'est écrit	aucun

Toutes ces clés sont rechargeables : le débit en vigueur s'applique à la requête suivante, ce qu'un principal a dépensé est gardé. Avec server.counters, les seaux sont partagés entre répliques comme ceux des limites HTTP : chaque réplique décide seule et se réconcilie toutes les sync (un débit tenu à une fenêtre sync près ; écart pas encore mesuré). Sans, chaque réplique compte pour elle.

Cas	Ce qui se passe
Client qui respecte KIP-219 (Java, franz-go)	Réponse immédiate avec `throttle_time_ms` ; il attend ce délai avant sa requête suivante
Client qui ignore le champ (librdkafka : confluent-kafka-go, Python, .NET)	Même réponse ; il renvoie aussitôt, mais sa requête suivante n'est servie qu'à l'échéance : le débit tient sans sa coopération
`Fetch` d'un principal ou d'un tenant déjà en dette	Réponse vide immédiate avec `throttle_time_ms`, rien n'est demandé au courtier
Courtier qui étrangle aussi la passerelle (ses propres quotas sur le principal de la passerelle)	`throttle_time_ms` = le plus grand des deux délais ; le silence du courtier est respecté : rien ne lui est envoyé sur cette connexion avant son terme
Quota jour ou mois du tenant épuisé	Refus `POLICY_VIOLATION` sur `Produce` (cause `quota`), pas un étranglement ; les lectures (`Fetch`) ne sont pas encore comptées au budget : leur code de refus doit d'abord être vérifié contre les principales bibliothèques clientes (un consommateur Java lève une exception sur un code inattendu)

Par tenant - un tenant ne prend pas toute la bande passante des courtiers, quel que soit son nombre de clés :

listeners:
  kafka:
    quotas:
      tenant: { producer_byte_rate: 10MiB, consumer_byte_rate: 40MiB, request_rate: 2000 }

Par client - chaque principal (jeton, clé, certificat) a son seau, sous le seau de son tenant :

listeners:
  kafka:
    quotas:
      client: { producer_byte_rate: 1MiB, consumer_byte_rate: 4MiB, request_rate: 200 }
      tenant: { producer_byte_rate: 10MiB, consumer_byte_rate: 40MiB }
      window: 2s           # des rafales de deux secondes de débit
      max_throttle: 5s     # des clients au request.timeout.ms court
server:
  counters: { store: "${COUNTERS_URL}" }   # le débit tenu sur toutes les répliques ensemble

Avec les quotas du courtier - rien à déclarer : la passerelle relaie le plus grand des deux délais. Le courtier voit le principal de la passerelle (upstreams.<nom>.sasl.username) et l'identifiant client du tenant (acme.app-1, Les noms des groupes et des transactions) ; un quota du courtier sur ce principal freine tous les tenants ensemble, ceux de la passerelle chacun le sien :

kafka-configs --bootstrap-server kafka-0.kafka:9093 --command-config admin.properties \
  --alter --entity-type users --entity-name flexway --add-config producer_byte_rate=104857600

Budget jour ou mois - le même forfait qu'en HTTP, compté sur les Produce :

listeners:
  kafka:
    quotas: { quota: plans }
server:
  counters: { store: "${COUNTERS_URL}" }   # exigé par toute politique quota
policies:
  definitions:
    plans: { type: quota, config: { limits: { day: 100000, month: 2000000 } } }

Un client librdkafka - rien à changer côté client ; ce qu'il voit au-delà du quota :

# confluent-kafka-go, Python, .NET : des propriétés ordinaires
bootstrap.servers=bootstrap.shop.kafka.example.com:9093
security.protocol=SASL_SSL
sasl.mechanisms=PLAIN
sasl.username=acme-orders-producer
sasl.password=${FLEXWAY_KAFKA_KEY}
# au-delà du quota : le producteur avance au débit déclaré ; sa statistique "throttle" (statistics.interval.ms)
# montre throttle_time_ms ; request.timeout.ms (30 s) reste au-dessus de max_throttle

Signaux : métriques, traces, audit

Les métriques flexway.kafka.* partent en OTLP avec les autres (même chemin, service.instance.id par réplique) ; chaque dimension est fermée : clusters du contrat, noms de la table des API, jeux ci-dessous, et tenant sous la règle de toutes les dimensions de tenant - 200 tenants gardent leur série, les suivants partagent _other. Jamais un sujet, une clé, un certificat, un client.id, un sujet Kafka physique ni une adresse.

Métrique	Dimensions	Ce qu'elle dit
`flexway.kafka.connections` (jauge)	`cluster`, `tenant`, `mechanism` (`oauthbearer`, `api_key`, `mtls`, `none` avant l'authentification ; `_none` et `_anonymous` avant le SNI)	Connexions ouvertes : sur une réplique, l'occupation de `max_connections` - qui ne refuse rien, une connexion de plus attend dans la file du noyau
`flexway.kafka.requests`	`cluster`, `tenant`, `api`, `decision` (`allowed`, `throttled`, `refused`), `cause` (`acl`, `admin_api`, `quota`, `unknown_topic`, `authentication`, `kms_unavailable`, `unsupported`, `key_destroyed` ; `none`)	Requêtes jugées ; `refused` dès qu'une partie est refusée par la passerelle
`flexway.kafka.bytes`	`cluster`, `tenant`, `direction` (`produce`, `fetch`)	Les octets que comptent les quotas
`flexway.kafka.throttle` (histogramme, s)	`cluster`, `tenant`, `quota` (`producer_byte_rate`, `consumer_byte_rate`, `request_rate`, `broker`), `scope` (`client`, `tenant`, `broker`)	`throttle_time_ms` de chaque réponse étranglée
`flexway.kafka.connections.refused`	`bound` (`max_unauthenticated_connections`, `max_connections_per_ip`, `max_connections_per_tenant`, `max_connection_rate_per_tenant`), `tenant` pour les deux bornes du tenant seulement	Connexions refusées par une borne ; jamais avant l'authentification un tenant, jamais une adresse ; le refus par `max_connections_per_tenant` n'est pas compté aussi en `authentication`
`flexway.kafka.connections.closed`	`cause` (`auth_timeout` sans tenant, `idle_timeout`, `max_session_lifetime`), `tenant`	Fermetures normales, jamais mêlées aux refus ; `max_session_lifetime` : session arrivée à son terme (la borne, ou `exp`, ou `NotAfter`) sans ré-authentification
`flexway.kafka.tenant_budget`	`cluster`, `tenant`, `outcome` (`waited`, `closed`)	Réponses qui ont attendu la part du tenant (`max_tenant_in_flight_bytes`, mode `wait`) ; connexions fermées à elle (mode `close`)
`flexway.kafka.tenant_budget.wait` (histogramme, s)	`cluster`, `tenant`	Combien de temps une réponse a attendu cette part
`flexway.kafka.encryption`	`cluster`, `tenant`, `op` (`encrypt`, `decrypt`), `outcome` (`ok`, `kms_unavailable`, `key_destroyed`)	Chiffrement par champ
`flexway.kafka.upstream.connections`	`cluster`, `broker` (identifiant de nœud, ou `bootstrap`), `outcome` (`ok`, `failed`)	Connexions de la passerelle à chaque courtier : la santé par courtier

Clé	Pourquoi	Défaut
`observability.kafka.budget_wait_buckets`	Bornes de `flexway.kafka.tenant_budget.wait`, croissantes, 20 au plus : à resserrer quand les attentes normales d'un parc tiennent sous la milliseconde. Non rechargeable (les bornes d'un histogramme se fixent à sa création)	`1ms`, `5ms`, `10ms`, `50ms`, `100ms`, `500ms`, `1s`, `5s`, `10s`, `30s` (le délai d'écriture)
`observability.kafka.closures`	Compter les fermetures normales : `false` sur un parc de clients éphémères où elles ne disent rien. Rechargeable	`true`

observability:
  kafka:
    budget_wait_buckets: [100us, 1ms, 10ms, 100ms, 1s, 10s, 30s]
    closures: false

Lire les signaux (noms Prometheus) :

# part des requêtes étranglées, par cluster virtuel
sum by (cluster) (rate(flexway_kafka_requests_total{decision="throttled"}[5m]))
  / sum by (cluster) (rate(flexway_kafka_requests_total[5m]))
# refus par cause (hors quota jour ou mois)
sum by (cause) (rate(flexway_kafka_requests_total{decision="refused", cause!="quota"}[5m]))
# connexions refusées par borne ; fermetures normales à part
sum by (bound) (rate(flexway_kafka_connections_refused_total[5m]))
sum by (cause) (rate(flexway_kafka_connections_closed_total[5m]))
# attente p95 de la part d'un tenant
histogram_quantile(0.95, sum by (le, tenant) (rate(flexway_kafka_tenant_budget_wait_seconds_bucket[5m])))

Traces : un span par requête authentifiée échantillonnée (kafka <API>, attributs kafka.cluster, tenant, auth.mechanism, kafka.decision, kafka.cause, kafka.throttle_time_ms) - jamais un sujet ni un nom de sujet Kafka ; hors échantillon, aucune allocation de plus (testing.AllocsPerRun).

Audit (audit.decisions : refusals ou all) : chaque requête refusée et chaque requête d'administration (permise ou refusée), dans la même chaîne que HTTP - host = cluster virtuel, method = KAFKA, operation = l'API, code = la cause, le sujet haché ; les requêtes de données permises sont comptées, pas journalisées une à une.

Dépannage

Symptôme côté client	Cause	Correction
`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

Déployer : le chart rend un Service <release>-flexway-kafka, sa NetworkPolicy et, avec cert-manager, le certificat à joker de chaque cluster (kafka.tls.clusters) - DEPLOYMENT. Quotas et signaux : Quotas : l'étranglement de KIP-219 et Signaux : métriques, traces, audit ; la console en Console : clusters virtuels, clients Kafka, règles, le chiffrement par champ en Chiffrement par champ.

Console : clusters virtuels, clients Kafka, règles

La console montre l'écouteur Kafka sans rubrique nouvelle :

Écran	Ce qu'il montre	Qui
Amonts, carte du courtier	Bloc « Proxy Kafka » : cluster virtuel, courtiers sains, connexions, tenants, sujets virtuels, octets produits et lus, part étranglée, refus par minute	Admins
Amonts, panneau du courtier	Cluster (adresse SNI, repli, identités admises, administration), courtiers physiques (connexions et échecs, jamais une adresse), sujets déclarés (forme physique `{tenant}.<nom>`, portées), connexions par tenant et par mode d'identité, débit d'octets, étranglement, refus par cause, rubrique « Limites », chiffrement par champ	Admins
Trafic, panneau « Clients Kafka »	Connexions par mode, requêtes par API et décision, octets, étranglement, refus par cause, budget du tenant, connexions refusées, lus à clé détruite, par cluster virtuel et par tenant	Admins ; un lecteur, son tenant seulement
Alertes	`kafka-proxy-refusals`, `kms-unavailable`, `kafka-tenant-budget`, `kafka-key-destroyed` (désactivée par défaut)	Admins

La rubrique Limites lit les refus de connexion par borne (flexway.kafka.connections.refused, max_unauthenticated_connections, max_connections_per_ip, max_connections_per_tenant, max_connection_rate_per_tenant), les fermetures normales à part (flexway.kafka.connections.closed : auth_timeout, idle_timeout, max_session_lifetime ; coupées par observability.kafka.closures: false), l'occupation de max_connections par la réplique la plus chargée (une jauge, jamais un refus), les attentes et fermetures du budget d'octets par tenant (flexway.kafka.tenant_budget, attente p95 sur flexway.kafka.tenant_budget.wait, seaux observability.kafka.budget_wait_buckets) et les bornes en vigueur de /config - lisibles même quand les métriques ne répondent pas.

Un lecteur ne voit que son tenant : ses attentes et fermetures, ses refus aux bornes de son tenant, jamais un refus d'avant l'authentification ni un autre tenant ; il voit les bornes de son tenant (max_connections_per_tenant, max_connection_rate_per_tenant, max_tenant_in_flight_bytes) et le mode on_tenant_budget_exceeded, sauf si l'exploitant les masque :

# valeurs du chart de la console
kafka:
  readerBounds: hide   # show par défaut ; FLEXWAY_CONSOLE_KAFKA_READER_BOUNDS

Règles intégrées, toutes réglables par builtins[] du fichier d'alertes (valeurs alerts.builtins du chart), sujets = clusters virtuels, jamais un tenant :

Règle	Défaut	Réglables
`kafka-proxy-refusals`	refus > 5 % des requêtes d'un cluster sur 5 min, dès 20 requêtes, toutes causes ; `for` 5 min	`threshold`, `minRequests`, `for`, `excludeCauses` (`acl`, `admin_api`, `quota`, `unknown_topic`, `authentication`, `kms_unavailable`, `unsupported`, `key_destroyed`), `enabled`
`kms-unavailable`	un échec `kms_unavailable` sur 1 min ; `for` 0 s ; une clé détruite ne la déclenche jamais	`for`, `enabled`
`kafka-tenant-budget`	attentes > 1 % des requêtes d'un cluster sur 5 min dès 20 requêtes, ou une fermeture sans minimum ; `for` 5 min ; valeur : la part et le nombre de tenants, jamais un nom	`threshold`, `minRequests`, `for`, `excludeCauses` (`waited`, `closed`), `enabled`
`kafka-key-destroyed`	désactivée ; activée : une issue `key_destroyed` (lecture `decrypt` ou écriture) sur 5 min, `for` 0 s	`enabled`, `for`

# valeurs du chart de la console
alerts:
  builtins:
    - { name: kafka-proxy-refusals, threshold: 10%, minRequests: 50, excludeCauses: [authentication, unknown_topic] }
    - { name: kms-unavailable, for: 1m }
    - { name: kafka-tenant-budget, threshold: 5%, excludeCauses: [closed] }   # fermetures ignorées
    - { name: kafka-key-destroyed, enabled: true }                            # effacement cryptographique signalé

enabled est un réglage de toute règle intégrée : une règle désactivée reste listée (« Désactivée »), n'est ni évaluée ni notifiée ; GET /api/alerts le dit dans tuning.enabled. Les compteurs du chiffrement restent à zéro tant qu'aucun sujet ne déclare de champ chiffré (Chiffrement par champ).

Clients Kafka natifs : clusters virtuels par tenant