Kafka in Production · Kafka Patterns and Operations
Transactions and Exactly-Once Processing
Read, transform and write with a transactional producer so output messages and consumer offsets commit atomically, and read only committed data.
What you'll build
A currency-conversion service that consumes payments.usd, writes converted amounts to payments.eur, and commits its input offsets in the same transaction as its output. An aborted transaction's messages are invisible to read_committed consumers.
The idea
- Kafka transactions let a producer write to several partitions and commit consumer offsets atomically (
send_offsets_to_transaction). - Consumers with
isolation.level=read_committednever see messages from aborted transactions. - Together this gives exactly-once processing within Kafka (consume, transform, produce). Side effects outside Kafka (databases, emails) still need idempotency.
- Each producer instance needs a stable, unique
transactional.id.
The code
import os
from confluent_kafka import Consumer, Producer, TopicPartition
from confluent_kafka.admin import AdminClient, NewTopic
BOOTSTRAP = os.environ["KAFKA_BOOTSTRAP"]
admin = AdminClient({"bootstrap.servers": BOOTSTRAP})
for f in admin.create_topics([NewTopic(t, 1, 1) for t in ("payments.usd", "payments.eur")]).values():
f.result()
seed = Producer({"bootstrap.servers": BOOTSTRAP})
for usd in (100, 250, 40):
seed.produce("payments.usd", value=str(usd))
seed.flush()
consumer = Consumer({"bootstrap.servers": BOOTSTRAP, "group.id": "fx-converter", "auto.offset.reset": "earliest",
"enable.auto.commit": False, "isolation.level": "read_committed"})
consumer.subscribe(["payments.usd"])
producer = Producer({"bootstrap.servers": BOOTSTRAP, "transactional.id": "fx-converter-1"})
producer.init_transactions()
for i in range(3):
msg = consumer.poll(10.0)
producer.begin_transaction()
producer.produce("payments.eur", value=str(round(int(msg.value()) * 0.92, 2)))
producer.send_offsets_to_transaction(consumer.position(consumer.assignment()), consumer.consumer_group_metadata())
if i == 1:
producer.abort_transaction() # e.g. a validation failure: output and offset both rolled back
print(f"aborted transaction for {msg.value().decode()} USD")
consumer.seek(TopicPartition(msg.topic(), msg.partition(), msg.offset())) # process this input again
msg = consumer.poll(10.0)
producer.begin_transaction()
producer.produce("payments.eur", value=str(round(int(msg.value()) * 0.92, 2)))
producer.send_offsets_to_transaction(consumer.position(consumer.assignment()), consumer.consumer_group_metadata())
producer.commit_transaction()
print(f"committed {msg.value().decode()} USD")
consumer.close()
reader = Consumer({"bootstrap.servers": BOOTSTRAP, "group.id": "eur-reader", "auto.offset.reset": "earliest",
"isolation.level": "read_committed"})
reader.subscribe(["payments.eur"])
values = []
while (m := reader.poll(5.0)) is not None:
values.append(m.value().decode())
reader.close()
print("read_committed consumer sees:", values)
for f in admin.delete_topics(["payments.usd", "payments.eur"]).values():
f.result()
for f in admin.delete_consumer_groups(["fx-converter", "eur-reader"]).values():
f.result()
Output
committed 100 USD aborted transaction for 250 USD committed 250 USD committed 40 USD read_committed consumer sees: ['92.0', '230.0', '36.8']
Run this on your computer — see “Get Started” for setup.
The aborted attempt's output was written to the log but is never shown to read_committed readers; each input was converted exactly once.
Common mistakes
- Expecting transactions to cover writes to databases or HTTP calls; they only cover Kafka topics and offsets.
- Sharing one
transactional.idbetween running instances: the newer one fences the older (by design). - Consumers left at the default
read_uncommitted, seeing aborted data. - Long transactions that block
read_committedconsumers (they can't read past an open transaction).
Key points
- Transactions commit produced messages and consumer offsets atomically.
- read_committed consumers ignore aborted messages.
- Exactly-once covers Kafka-to-Kafka; external effects still need idempotency.
Further reading: Kafka transactions and exactly-once semantics · KIP-98: exactly once delivery
Practice questions
What does send_offsets_to_transaction achieve?
Answer
It commits the consumer's input offsets as part of the producer's transaction, so outputs and progress are committed (or aborted) together.
What does isolation.level=read_committed change for consumers?
Answer
They only see messages from committed transactions, never from aborted ones.
Does Kafka exactly-once cover writing to a database?
Answer
No; it covers Kafka topics and offsets. External side effects need idempotency or an outbox pattern.
Open in the app to mark lessons complete, take notes and track your progress.