# Clients Kafka natifs

> Un producteur, un consommateur ou un outil Kafka de n'importe quelle bibliothèque se branche sur la passerelle sans changer de code. Disponible depuis 1.6.0…

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

> Un producteur, un consommateur ou un outil Kafka de n'importe quelle bibliothèque se branche sur la passerelle sans changer de code. Disponible depuis `1.6.0`.
>
> Côté produit : [Clients Kafka natifs et chiffrement par champ](https://flexway.io/docs/product-kafka/) ; côté exploitation : [Clients Kafka natifs : clusters virtuels par tenant](https://flexway.io/docs/operations-kafka-clients/), [Chiffrement par champ](https://flexway.io/docs/operations-kafka-encryption/).

## Se connecter

| Propriété | Valeur |
| --- | --- |
| `bootstrap.servers` | `bootstrap.<cluster>.<domaine>:9093`, donné par l'exploitant |
| Chiffrement | TLS |
| Identité | `OAUTHBEARER` (un jeton de l'émetteur déclaré), `PLAIN` (identifiant de la clé d'API en utilisateur, la clé en mot de passe) ou un certificat client |
| `group.protocol` | `classic` (le défaut) : le protocole de KIP-848 n'est pas servi |

```properties
# librdkafka (confluent-kafka-go, Python, .NET) - un jeton OAUTHBEARER
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
```

```properties
# Client Java - 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}";
```

```go
// franz-go - une clé d'API en PLAIN
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()),
)
```

## Ce que vous voyez

Votre **cluster virtuel** : les seuls topics que le contrat vous ouvre, sous leurs noms courts. Vos
groupes et vos identifiants de transaction sont rangés par la passerelle dans l'espace de votre
tenant, sans que vous les changiez. La passerelle présente ses propres identifiants au courtier.

| Code Kafka | Sens | Que faire |
| --- | --- | --- |
| `TOPIC_AUTHORIZATION_FAILED`, `GROUP_AUTHORIZATION_FAILED`, `TRANSACTIONAL_ID_AUTHORIZATION_FAILED` | Opération non ouverte à votre identité | Demander l'accès |
| `UNKNOWN_TOPIC_OR_PARTITION` | Topic hors de votre cluster virtuel | Vérifier le nom court |
| `SASL_AUTHENTICATION_FAILED` | Identifiant refusé ou révoqué | Renouveler la clé ou le jeton |
| `POLICY_VIOLATION` (44) | Quota du jour ou du mois dépensé (sur une écriture seulement : les lectures ne sont pas encore comptées), ou administration refusée | Attendre la fenêtre suivante |
| `INVALID_RECORD` (87) | En-tête `flexway-*` écrit par vous, valeur que son schéma ne décode pas | Corriger l'enregistrement |
| `KAFKA_STORAGE_ERROR` (56) | KMS indisponible : rien n'est écrit en clair | Le client rejoue seul |
| `RESOURCE_NOT_FOUND` (91) | Clé de chiffrement détruite | Ne pas rejouer |

Une connexion est fermée quand votre identifiant est révoqué ou à la fin de la durée de session : le
client se reconnecte et se réauthentifie seul (KIP-368).

## Quotas : `throttle_time_ms`

Au-delà de votre débit, la réponse arrive **tout de suite** avec `throttle_time_ms`, et la passerelle
ne lit plus votre connexion pendant ce délai. Un client Java ou franz-go attend de lui-même ;
librdkafka ignore le champ mais est freiné de même. Gardez `request.timeout.ms` (30 s par défaut)
au-dessus du plafond d'étranglement que l'exploitant annonce.

## Champs chiffrés

Les champs que le contrat déclare sensibles sont chiffrés par la passerelle. Un consommateur dont le
jeton porte la **portée de déchiffrement** les reçoit en clair ; les autres reçoivent la valeur
chiffrée (en base64 dans un champ texte). Si la clé a été détruite (effacement voulu), l'enregistrement
arrive avec le champ **toujours chiffré** et l'en-tête `flexway-encryption: key_destroyed` - lisez-le
avant d'utiliser le champ :

```python
for msg in consumer.consume(100, timeout=1.0):
    headers = dict(msg.headers() or [])
    if headers.get("flexway-encryption") == b"key_destroyed":
        continue  # effacement voulu : ne pas utiliser les champs chiffrés
    handle(msg.value())
```

Un champ altéré produit `CORRUPT_MESSAGE` (2) sur la partition, jamais une valeur.