flexway
FRENParler à un expert

Clients Kafka natifs

GuideDéveloppeursDepuis la version 1.6.0

Markdown

Sur cette page

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 ; côté exploitation : Clients Kafka natifs : clusters virtuels par tenant, Chiffrement par champ.

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 
# 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
# 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}";
// 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 :

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.

Étapes suivantes