Vue d'ensemble

confluent-kafka 2.2, publié le 10 juillet 2023, améliore l'API transactionnelle pour les opérations exactly-once.

Fonctionnalités principales

API transactionnelle

L'API transactionnelle permet de produire et consommer des messages dans une transaction atomique, garantissant la sémantique exactly-once.

python 
from confluent_kafka import Producer

conf = {
    'bootstrap.servers': 'localhost:9092',
    'transactional.id': 'mon-tx-id',
}

producer = Producer(conf)
producer.init_transactions()
producer.begin_transaction()
try:
    producer.produce('topic-a', value='message-1')
    producer.produce('topic-b', value='message-2')
    producer.commit_transaction()
except Exception:
    producer.abort_transaction()

Améliorations de fiabilité

Le client bénéficie de correctifs de fiabilité pour la reconnexion automatique et la gestion des erreurs transitoires.

python 
from confluent_kafka import Consumer

consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'mon-groupe',
    'enable.auto.commit': False,
    'isolation.level': 'read_committed',
})
consumer.subscribe(['topic-a'])

msg = consumer.poll(timeout=5.0)
if msg and not msg.error():
    print(f'Reçu : {msg.value().decode()}')
    consumer.commit(msg)
consumer.close()

Sources