Kafka Consumers and Consumer Groups

A consumer reads from partitions. A consumer group is a set of consumers that divide those partitions between them, so adding a consumer adds throughput. That division, and what happens when it changes, is most of what you need to understand.

Reading records from Python

from confluent_kafka import Consumer, KafkaError

consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "demo-group",
    "auto.offset.reset": "earliest",
})
consumer.subscribe(["orders"])

try:
    while True:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() != KafkaError._PARTITION_EOF:
                print(f"error: {msg.error()}")
            continue

        print(f"p{msg.partition()} @{msg.offset()}  "
              f"{msg.key().decode() if msg.key() else '-':10} {msg.value().decode()}")
finally:
    consumer.close()

$ python consume.py
p0 @0      order-1    espresso
p0 @1      order-2    cortado
p0 @2      order-3    flat white
p0 @4      counter    espresso
p0 @5      counter    cortado
p0 @6      counter    flat white

auto.offset.reset: earliest is what made it start at offset 0. That setting only applies when the group has no committed offset for a partition — it is the starting position for a brand-new group, not a “always read from the beginning” switch.

Stop the consumer and run it again:

$ python consume.py

Nothing, and this is correct. The group demo-group now has committed offsets, so it resumes where it stopped rather than replaying. Change group.id to demo-group-2 and all the records come back — a new group starts fresh.

Two consumers, one group

Give the topic more partitions so there is work to divide, then run the same script twice in two terminals. Add a print of the assignment so you can see what each one owns:

def on_assign(consumer, partitions):
    owned = sorted(p.partition for p in partitions)
    print(f"assigned partitions: {owned}")

consumer.subscribe(["events"], on_assign=on_assign)

Terminal 1, started first:

$ python consume.py
assigned partitions: [0, 1, 2]
p0 @0      -          event-1
p1 @0      -          event-2
p2 @0      -          event-3

One consumer, so it owns all three partitions. Now start terminal 2:

$ python consume.py
assigned partitions: [2]

And terminal 1 prints:

assigned partitions: [0, 1]

The group rebalanced. Kafka revoked all assignments and redistributed them: consumer 1 kept partitions 0 and 1, consumer 2 got partition 2. Neither consumer chose this — the group coordinator on the broker decided and pushed the assignment out.

Start a third consumer and each gets one partition. Start a fourth:

$ python consume.py
assigned partitions: []

Empty. There are only three partitions and they are all taken. The fourth consumer sits idle until one of the others dies, at which point it inherits that partition. This is the hard ceiling on consumer-group parallelism: partition count is the maximum useful consumer count.

Two groups, same topic

Change group.id to analytics in one terminal while demo-group keeps running:

$ python consume.py            # group.id=analytics
assigned partitions: [0, 1, 2]
p0 @0      -          event-1
p1 @0      -          event-2
p2 @0      -          event-3

It gets all three partitions and every record, while demo-group continues independently. Groups do not compete; each maintains its own offsets and receives a complete copy of the stream. This is how one topic feeds a billing service, a search indexer, and an audit log without any of them affecting the others.

Auto-commit will lose your work

The default enable.auto.commit=true commits the current position every 5 seconds, on a timer, regardless of whether your processing succeeded. Simulate a crash:

import time
from confluent_kafka import Consumer

consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "fragile",
    "auto.offset.reset": "earliest",
})
consumer.subscribe(["orders"])

processed = 0
while True:
    msg = consumer.poll(1.0)
    if msg is None or msg.error():
        continue

    print(f"received @{msg.offset()}")
    time.sleep(3)                     # slow work
    if msg.offset() == 2:
        raise RuntimeError("crash during processing of offset 2")
    processed += 1
    print(f"  done @{msg.offset()}")

$ python fragile.py
received @0
  done @0
received @1
  done @1
received @2
Traceback (most recent call last):
  ...
RuntimeError: crash during processing of offset 2

Restart it:

$ python fragile.py
received @3

Offset 2 was never processed and is never redelivered. The auto-commit timer fired during the three-second sleep and committed position 3, so the group believes offset 2 is done. The record is still in the log, but this group will never read it again.

Manual commits fix it

consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "careful",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,
})
consumer.subscribe(["orders"])

while True:
    msg = consumer.poll(1.0)
    if msg is None or msg.error():
        continue

    print(f"received @{msg.offset()}")
    time.sleep(3)
    if msg.offset() == 2 and not os.path.exists("/tmp/retried"):
        open("/tmp/retried", "w").close()
        raise RuntimeError("crash during processing of offset 2")

    consumer.commit(msg)              # commit only after the work succeeded
    print(f"  done and committed @{msg.offset()}")

$ python careful.py
received @0
  done and committed @0
received @1
  done and committed @1
received @2
Traceback (most recent call last):
  ...
RuntimeError: crash during processing of offset 2

$ python careful.py
received @2
  done and committed @2
received @3
  done and committed @3

Offset 2 is redelivered and processed. Committing after the work rather than on a timer converts silent loss into a retry — this is at-least-once delivery, and it means your processing must tolerate seeing the same record twice.

Practice

assigned partitions: [0, 1]
# ... other consumer killed ...
assigned partitions: [0, 1, 2, 3]

received @0
%4|...|MAXPOLL|rdkafka#consumer-1| [thrd:main]: Application maximum poll interval (5000ms) exceeded by 5013ms
assigned partitions: []

bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group demo-group --topic orders --reset-offsets --to-earliest --execute

GROUP                          TOPIC          PARTITION  NEW-OFFSET
demo-group                     orders         0          0

bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group demo-group

GROUP        TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID
demo-group   events  0          12              12              0    rdkafka-8f2a...
demo-group   events  1          5               7               2    rdkafka-8f2a...
demo-group   events  2          8               8               0    rdkafka-3c91...

Next: how keys decide partitioning, and what ordering you can actually rely on.