# Clients Kafka natifs : clusters virtuels par tenant

> Les producteurs, consommateurs et outils d'administration Kafka - n'importe quelle bibliothèque : franz-go, librdkafka (confluent-kafka-go, Python, .NET), le…

Guide - Depuis la version 1.6.0 - Flexway v1.6.3 - https://flexway.io/docs/operations-kafka-clients/

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

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](#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](#quotas--létranglement-de-kip-219)) ;
> - 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.

<!-- example: examples/kafka-proxy/flexway.yaml -->

```yaml
# 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
```

```yaml
# 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](https://flexway.io/docs/operations-api-keys/)) : 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)](https://flexway.io/docs/operations-client-certificates/)), 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](#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).

```yaml
# 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 :

```yaml
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 :

```go
// 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 }))
```

```properties
# 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
```

```properties
# 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](#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 |

```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é
```

## 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).

```yaml
# 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 } } } ]
```

```yaml
# 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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-219+-+Improve+quota+communication)). Un client qui ignore le
champ - librdkafka ne l'a jamais appliqué ([librdkafka#5592](https://github.com/confluentinc/librdkafka/issues/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](https://flexway.io/docs/operations-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 :

```yaml
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 :

```yaml
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](#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 :

```bash
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` :

```yaml
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 :

```properties
# 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](https://flexway.io/docs/operations-kafka-encryption/) |
| `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` |

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

**Lire les signaux** (noms Prometheus) :

```promql
# 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](#administration-bornée) ; pour le budget, la fenêtre suivante (UTC) ou une politique `quota` relevée ([Quotas : l'étranglement de KIP-219](#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](#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](https://flexway.io/docs/operations-kafka-encryption/), 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](#quotas--létranglement-de-kip-219) et
[Signaux : métriques, traces, audit](#signaux--métriques-traces-audit) ; la console en [Console : clusters virtuels, clients Kafka, règles](#console--clusters-virtuels-clients-kafka-règles), le chiffrement par champ en
[Chiffrement par champ](https://flexway.io/docs/operations-kafka-encryption/).

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

```yaml
# 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` |

```yaml
# 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](https://flexway.io/docs/operations-kafka-encryption/)).