The Consumer API

The Consumer API

The Consumer API and the Poll Loop

A consumer subscribes to topics and calls poll() in a loop. Behind it, the client fetches batches from each partition's leader into a local queue; poll() hands out records from that queue and also runs the group's callbacks. Kafka 129 assumes a member that has not polled for max.poll.interval.ms (5 minutes) is stuck and gives its partitions away, so keep the work per poll short or hand it to other threads.

l0671_poll.py: a minimal poll loopPython

from confluent_kafka import Consumer, KafkaException
consumer = Consumer({"bootstrap.servers": "localhost:33092", "group.id": "booknest-first-look",
                     "auto.offset.reset": "earliest"})   # where to start without a commit
consumer.subscribe(["booknest.order-events"])
seen = 0
try:
    while seen < 1000:
        msg = consumer.poll(1.0)              # fetches in the background, returns one record
        if msg is None:
            continue                          # nothing yet: poll again
        if msg.error():
            raise KafkaException(msg.error())
        seen += 1
        if seen <= 2:
            print(msg.partition(), msg.offset(), msg.key(), msg.value()[:50])
    print("next offsets:", [tp.offset for tp in consumer.position(consumer.assignment())])
finally:
    consumer.close()                          # commits final offsets and leaves the group

Output

2 0 b'2' b'{"event_id":2,"ts":"2025-01-01T00:04:32Z","type":"'
2 1 b'3' b'{"event_id":3,"ts":"2025-01-01T00:06:13Z","type":"'
next offsets: [-1001, -1001, 1000]

The first thousand records all came from partition 2, so the others have no position yet (-1001). Without a committed offset a group starts where auto.offset.reset says (default latest). Always close() a consumer: it commits final offsets and leaves the group at once rather than after a session timeout.

MENU