Transform a stream

The short version

A stream processor is a consumer and a producer in one loop: read a message, transform it, write the result to another topic, then commit.

Details

Write the output before you commit the input. If the processor crashes in between, it repeats the work instead of losing it. In Java, Kafka Streams runs this loop for you. The script below shows what it does underneath.

from confluent_kafka import Consumer, Producer

conf = {"bootstrap.servers": "localhost:9092"}
consumer = Consumer({**conf, "group.id": "ticketer",
    "auto.offset.reset": "earliest", "enable.auto.commit": False})
producer = Producer({**conf, "acks": "all"})
consumer.subscribe(["orders"])

while True:
    msg = consumer.poll(1.0)
    if msg is None or msg.error():
        continue
    producer.produce("tickets", key=msg.key(), value=msg.value().upper())
    producer.flush()
    consumer.commit(message=msg, asynchronous=False)

Try it yourself

Create tickets with 3 partitions, save the code as ticketer.py and run it. Then read tickets with the console consumer and --from-beginning.

You should see every order in uppercase, such as PIZZA and CAKE.

Explain it to the duck

Why must the processor write the result before it commits the input?

Tested with: Kafka 4.1.0 in Docker, Python 3.12 and confluent-kafka 2.11.