Kafka — Producers & Consumers in Python¶

uv add confluent-kafka                              # producer, consumer, admin
uv add "confluent-kafka[avro,schemaregistry]"       # + Avro serializers and Schema Registry client

from confluent_kafka.admin import AdminClient, NewTopic

admin = AdminClient({"bootstrap.servers": "localhost:9092"})

futures = admin.create_topics([
    NewTopic("orders", num_partitions=3, replication_factor=1, config={"retention.ms": "604800000"}),
    NewTopic("orders.dlq", num_partitions=1, replication_factor=1),
])
for topic, future in futures.items():
    try:
        future.result()          # raises KafkaException, e.g. TOPIC_ALREADY_EXISTS
        print(f"created {topic}")
    except Exception as exc:
        print(f"{topic}: {exc}")

print(sorted(t for t in admin.list_topics(timeout=5).topics if not t.startswith("__")))

import json
import socket

from confluent_kafka import KafkaError, Message, Producer

producer = Producer({
    "bootstrap.servers": "localhost:9092",
    "client.id": f"orders-api-{socket.gethostname()}",
    "acks": "all",
    "enable.idempotence": True,
    "linger.ms": 5,
    "compression.type": "zstd",
    "delivery.timeout.ms": 30000,
})


def on_delivery(err: KafkaError | None, msg: Message) -> None:
    if err is not None:
        print(f"FAILED {msg.key()}: {err}")          # real code: log, metric, alert
    else:
        print(f"ok {msg.key().decode()} -> {msg.topic()}[{msg.partition()}]@{msg.offset()}")


for order_id, status in [("ord-1", "created"), ("ord-2", "created"), ("ord-1", "paid")]:
    producer.produce(
        "orders",
        key=order_id,                                  # str or bytes; str is UTF-8 encoded
        value=json.dumps({"order_id": order_id, "status": status}).encode(),
        headers={"event-type": "order-status-changed", "schema-version": "1"},
        on_delivery=on_delivery,
    )
    producer.poll(0)                                   # serve callbacks of earlier messages

remaining = producer.flush(10)                         # wait for all outstanding deliveries
assert remaining == 0, f"{remaining} messages not delivered"

import json
import signal

from confluent_kafka import Consumer, KafkaException

running = True


def stop(signum, frame):
    global running
    running = False


signal.signal(signal.SIGTERM, stop)                 # Kubernetes / docker stop

consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "billing",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": True,
    "enable.auto.offset.store": False,              # we decide what is done
})
consumer.subscribe(["orders"])


def handle(event: dict) -> None:
    print("processing", event)


try:
    while running:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            raise KafkaException(msg.error())
        handle(json.loads(msg.value()))
        consumer.store_offsets(message=msg)         # committed by the next auto commit
finally:
    consumer.close()                                # commits stored offsets and leaves the group

from confluent_kafka import Consumer


def on_assign(consumer, partitions):
    print("assigned", [(p.topic, p.partition) for p in partitions])


def on_revoke(consumer, partitions):
    print("revoked", [(p.topic, p.partition) for p in partitions])
    # manual commits: commit processed offsets here, before another member takes over


consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "billing",
    "group.protocol": "consumer",        # KIP-848: broker-side, incremental assignment
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,
})
consumer.subscribe(["orders"], on_assign=on_assign, on_revoke=on_revoke)

import json

from confluent_kafka import Consumer, KafkaException, Producer

consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "invoicer",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,
    "isolation.level": "read_committed",
})
producer = Producer({
    "bootstrap.servers": "localhost:9092",
    "transactional.id": "invoicer-1",          # stable per instance; enables idempotence
})
producer.init_transactions()
consumer.subscribe(["orders"])

while True:
    msgs = consumer.consume(num_messages=100, timeout=1.0)
    if not msgs:
        continue
    producer.begin_transaction()
    try:
        for msg in msgs:
            if msg.error():
                raise KafkaException(msg.error())
            order = json.loads(msg.value())
            producer.produce("invoices", key=msg.key(), value=json.dumps({"invoice_for": order["order_id"]}).encode())
        producer.send_offsets_to_transaction(            # input offsets commit together with the output
            consumer.position(consumer.assignment()),
            consumer.consumer_group_metadata(),
        )
        producer.commit_transaction()
    except Exception:
        producer.abort_transaction()
        raise                                            # real code: seek back to the committed offsets and retry

import asyncio

from confluent_kafka.aio import AIOConsumer, AIOProducer


async def main() -> None:
    producer = AIOProducer({"bootstrap.servers": "localhost:9092"})
    try:
        delivery = await producer.produce("orders", key="ord-9", value=b'{"order_id": "ord-9"}')
        msg = await delivery                        # resolves on the delivery report
        print("delivered", msg.partition(), msg.offset())
    finally:
        await producer.flush()
        await producer.close()

    consumer = AIOConsumer({
        "bootstrap.servers": "localhost:9092",
        "group.id": "aio-demo",
        "auto.offset.reset": "earliest",
    })
    try:
        await consumer.subscribe(["orders"])
        for _ in range(10):
            msg = await consumer.poll(1.0)
            if msg is not None and not msg.error():
                print("consumed", msg.key(), msg.offset())
    finally:
        await consumer.close()


asyncio.run(main())