The Poll Loop

Concepts covered: Consumer groups

The consumer pulls

A Kafka consumer pulls. It sends a fetch request to the leader of each partition it reads, naming the offset it wants next, and the leader answers with the records from that offset forward. The call that does this is poll. One call can return hundreds of records from several partitions, or none at all when nothing new has arrived, and the consumer decides when to call again. The leader tracks nothing about what the consumer does with those records; it answers fetches. A broker acting as the group coordinator keeps track of which consumers read together and where each one has got to, and sections 1 and 2 cover it.

The log lesson showed that a record sits at an offset in a partition and that reading leaves it there. A consumer turns that into a loop: subscribe to a topic, call poll, handle every record in the batch, call poll again. The loop below is the whole job of a billing service that reads the orders topic and charges a card for each one.

consumer.subscribe(["orders"])

while True:
    batch = consumer.poll(timeout_ms=1000)
    for partition, records in batch.items():
        for record in records:
            charge_card(record.key, record.value)
    consumer.commit()

Suppose the loop's 2nd poll returns 3 records. The consumer did not ask for offsets 4 and 5 of partition 0 by name; it asked for everything from its current position on each partition it holds, and the 2 leaders sent what they had. Within 1 partition the batch keeps offset order, 4 before 5. Across partitions there is no order, and partition 1's record may arrive before or after partition 0's.

After that batch the consumer remembers, in its own memory, that partition 0 is at 6 and partition 1 is at 13, and the next poll fetches from those positions. The position is in the consumer's memory alone; nothing has been saved to the broker. Saving it is the commit, which section 2 covers.

What one poll may return

max.poll.records caps how many records a single poll hands to your code, 500 by default; the consumer fetches more than that in the background and parcels them out call by call. fetch.min.bytes and fetch.max.wait.ms tell the leader how long it may hold a fetch open while data piles up before it answers. With the defaults of 1 byte and 500 ms, the leader answers as soon as it has anything. max.partition.fetch.bytes limits how much of 1 partition comes back in 1 fetch, so a batch of 1 KB records from 1 partition tops out near 1,000 records in the buffer.

The deadline between polls

Between one poll and the next, the consumer does its work on the batch: it charges the cards, writes the rows, calls the downstream service. The broker does not watch this work; it sees only the next fetch. The consumer must call poll again within max.poll.interval.ms, 5 minutes by default. A batch of 500 records that take 1 second each finishes in about 8 minutes. That is past the deadline, so the group treats the consumer as gone. What the group does about it comes in a later lesson; for now, size the batch so the loop returns in time: 500 records at 1 second each needs max.poll.records under 300, or a handler 2 times faster.

Watch the consumer's position in the animation. After each batch it moves forward by the batch size, and the next fetch asks for exactly that offset. The setting changes how many records the consumer holds at once between polls, and nothing about which records it reads or in what order.

About this lesson

This section is part of the Consumer Groups: Beginner lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.