Kafka in Production · Kafka Fundamentals for Production
Consumer Groups and Rebalancing
Scale consumption with consumer groups, see how partitions are assigned and rebalanced, and keep processing within poll intervals.
What you'll build
Two consumers in the same group sharing a three-partition topic: each gets a subset of partitions, together they read every message exactly once per group, and a second group independently reads the same messages again.
The idea
- Consumers with the same
group.idform a consumer group; Kafka assigns each partition to one member. - When members join or leave (deploys, crashes), partitions are rebalanced.
- Different groups are independent: each has its own offsets, so several services can read the same topic.
- Consumers must call
poll()regularly; if processing takes longer thanmax.poll.interval.ms, the member is removed from the group.
The code
import os
import threading
import time
from confluent_kafka import Consumer, Producer
from confluent_kafka.admin import AdminClient, NewTopic
BOOTSTRAP = os.environ["KAFKA_BOOTSTRAP"]
admin = AdminClient({"bootstrap.servers": BOOTSTRAP})
admin.create_topics([NewTopic("clicks", num_partitions=3, replication_factor=1)])["clicks"].result()
producer = Producer({"bootstrap.servers": BOOTSTRAP})
for i in range(30):
producer.produce("clicks", key=f"user-{i % 7}", value=f"click {i}")
producer.flush()
def consume(group, name, results, seconds=8):
consumer = Consumer({"bootstrap.servers": BOOTSTRAP, "group.id": group, "auto.offset.reset": "earliest"})
consumer.subscribe(["clicks"])
deadline = time.time() + seconds
while time.time() < deadline:
msg = consumer.poll(0.5)
if msg is not None and not msg.error():
results.setdefault(name, set()).add((msg.partition(), msg.offset()))
consumer.close()
analytics = {}
workers = [threading.Thread(target=consume, args=("analytics", n, analytics)) for n in ("a", "b")]
for w in workers: w.start()
for w in workers: w.join()
partitions_per_member = sorted(len({p for p, _ in msgs}) for msgs in analytics.values())
total = sum(len(m) for m in analytics.values())
print("analytics group: partitions per member", partitions_per_member, "| messages", total)
audit = {}
consume("audit", "only", audit, seconds=5)
print("audit group read the same topic again:", len(audit["only"]), "messages")
admin.delete_topics(["clicks"])["clicks"].result()
for f in admin.delete_consumer_groups(["analytics", "audit"]).values():
f.result()
Output
analytics group: partitions per member [1, 2] | messages 30 audit group read the same topic again: 30 messages
Run this on your computer — see “Get Started” for setup.
Common mistakes
- More consumers than partitions: the extra ones sit idle.
- Slow processing inside the poll loop (calling slow APIs per message) exceeding
max.poll.interval.ms, causing constant rebalances. - Two unrelated services sharing a
group.idand stealing each other's messages. - Doing work in partition-revocation callbacks without committing what's been processed.
Key points
- A consumer group splits partitions among its members.
- Different groups read the same data independently.
- Poll regularly; long processing triggers rebalances.
Further reading: Consumer configs · Consumer group protocol (KIP-848)
Practice questions
What happens if a consumer group has more members than partitions?
Answer
Each partition goes to one member, so the extra members get no partitions and stay idle.
How can two services both process every message of a topic?
Answer
Give them different group.id values; each group keeps its own offsets.
What does max.poll.interval.ms control?
Answer
The maximum time between poll() calls before the consumer is considered failed and its partitions are rebalanced.
Open in the app to mark lessons complete, take notes and track your progress.