TopolearnEvery Kafka-driven service is built around a consumer loop, and its small details decide correctness.
A consumer needs bootstrap.servers and a group.id (the consumer group, covered next). It subscribes to topics and then repeatedly calls poll(timeout), which returns one record or None if nothing arrived in time. Kafka consumers pull: the broker never pushes data, so a slow consumer just falls behind instead of being overwhelmed.
Each returned message may carry an error (for example, end-of-partition events or a deleted topic); check msg.error() before using it. msg.key(), msg.value(), msg.partition(), msg.offset() and msg.timestamp() give the record's contents and position.
Where does a brand-new group start? auto.offset.reset decides when there is no committed offset: latest (the default, only new records) or earliest (from the oldest retained record). Once the group has committed offsets, it resumes from them regardless of this setting.
By default the client auto-commits offsets periodically in the background (every 5 s in librdkafka and Java), which is simple but can lose or repeat records on crashes; later lessons cover manual commits.
Always call close() on shutdown: it commits final offsets and leaves the group immediately, so its partitions are reassigned without waiting for a session timeout. For batch work, consume(num_messages, timeout) returns a list.
Example
import json
from confluent_kafka import Consumer
consumer = Consumer({
"bootstrap.servers": "localhost:9092",
"group.id": "billing",
"auto.offset.reset": "earliest", # new group: start from the oldest record
})
consumer.subscribe(["orders"])
try:
while True:
msg = consumer.poll(1.0)
if msg is None:
continue # nothing within 1 s
if msg.error():
print("error:", msg.error())
continue
event = json.loads(msg.value())
print(msg.partition(), msg.offset(), msg.key().decode(), event["status"])
# 1 0 c1 created
except KeyboardInterrupt:
pass
finally:
consumer.close() # commit and leave the group cleanlyRemember
- Subscribe, then poll in a loop; consumers pull at their own pace
- Check msg.error() before using a message
auto.offset.resetonly applies when the group has no committed offset- Always close() to commit and leave the group
Watch out for
- Changing auto.offset.reset to earliest for an existing group and expecting it to re-read history; committed offsets win.