Kafka Producers, Consumers and Consumer Groups Explained

A courier-dispatch service reads delivery requests from Kafka. At the lunch rush it falls behind, so someone starts a second copy of the service. If both copies used different group ids, every request would be dispatched twice. If they share a group id but the topic has one partition, the second copy would do nothing. Consumer groups are the mechanism behind both outcomes, and understanding them is what separates "I can send and receive a message" from "I can scale a consumer".

The producer side

A producer does not send one network request per record. It places records in per-partition batches in memory and ships a batch when it is full or when a short delay passes. Three settings govern the trade-off between latency, throughput and safety.

Here is a producer written with the confluent-kafka Python library. It was not executed for this lesson because there is no broker here, so check the library version's documentation before copying it, but the call shapes are the standard ones: produce is asynchronous, a delivery callback reports success or failure later, and flush blocks until everything is acknowledged.

import json
from confluent_kafka import Producer

producer = Producer({
    "bootstrap.servers": "localhost:9092",
    "acks": "all",
    "linger.ms": 5,
    "compression.type": "lz4",
})

def on_delivery(err, msg):
    if err is not None:
        print("delivery failed:", err)
    else:
        print(f"stored in {msg.topic()} partition {msg.partition()} offset {msg.offset()}")

request = {"order_id": "A-1042", "pickup": "Baner", "drop": "Kothrud"}
producer.produce(
    "delivery-requests",
    key=request["order_id"],
    value=json.dumps(request),
    on_delivery=on_delivery,
)
producer.poll(0)   # serve delivery callbacks
producer.flush()   # wait for outstanding acks before the program exits

The call to produce returning does not mean the record is stored; it means the record was queued locally. Skipping flush() at shutdown is how records silently vanish from short scripts.

Consumer groups: dividing the work

Every consumer belongs to a group identified by group.id. Kafka's rule is simple: within one group, each partition is assigned to exactly one consumer at a time. Different groups are independent and each receives every record, which is how the billing and analytics services share one topic. Scaling a service means starting more consumers with the same group id, up to the number of partitions, a horizontal-scaling pattern covered more generally in Scalability and Load Balancing. The sketch below models the assignment, using made-up offsets to compute lag, which is the distance between the end of the log and what the group has committed.

def assign(partitions, members):
    """Spread partitions over members round-robin (a simplified assignor)."""
    plan = {m: [] for m in sorted(members)}
    names = sorted(members)
    for i, p in enumerate(partitions):
        if names:
            plan[names[i % len(names)]].append(p)
    return plan

partitions = list(range(6))
log_end = {0: 120, 1: 118, 2: 140, 3: 90, 4: 101, 5: 99}   # next offset to be written
committed = {0: 100, 1: 100, 2: 100, 3: 90, 4: 90, 5: 99}  # group's committed offsets

def show(label, members):
    print(label)
    plan = assign(partitions, members)
    for m, ps in plan.items():
        lag = sum(log_end[p] - committed[p] for p in ps)
        print(f"  {m}: partitions {ps} lag={lag}")
    if not plan:
        print("  (no members)")

show("3 consumers in group 'dispatch':", {"c-a", "c-b", "c-c"})
show("c-b crashes -> rebalance:", {"c-a", "c-c"})
show("10 consumers, 6 partitions:", {f"c-{i}" for i in range(10)})

3 consumers in group 'dispatch':
  c-a: partitions [0, 3] lag=20
  c-b: partitions [1, 4] lag=29
  c-c: partitions [2, 5] lag=40
c-b crashes -> rebalance:
  c-a: partitions [0, 2, 4] lag=71
  c-c: partitions [1, 3, 5] lag=18
10 consumers, 6 partitions:
  c-0: partitions [0] lag=20
  c-1: partitions [1] lag=18
  c-2: partitions [2] lag=40
  c-3: partitions [3] lag=0
  c-4: partitions [4] lag=11
  c-5: partitions [5] lag=0
  c-6: partitions [] lag=0
  c-7: partitions [] lag=0
  c-8: partitions [] lag=0
  c-9: partitions [] lag=0

Real assignors (range, sticky, cooperative sticky) use different rules, so which partitions go to which consumer will differ, but the invariants are the same. The ceiling on parallelism is the partition count: ten consumers on six partitions leave four idle. When a member leaves, its partitions move to the survivors, which is the rebalance, and the survivors' lag jumps because they now carry more.

Commits: how a group remembers its place

A consumer's progress is stored in Kafka itself, in an internal topic called __consumer_offsets, keyed by group, topic and partition. By convention the committed value is the offset of the next record to read, so after fully handling offset 41 you commit 42. By default the client commits automatically on a timer; that is convenient and also the source of lost or repeated work, because the timer knows nothing about whether your processing finished. For anything with side effects, turn auto-commit off and commit after processing.

import json
from confluent_kafka import Consumer

consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "dispatch",
    "enable.auto.commit": False,
    "auto.offset.reset": "earliest",   # only used when the group has no committed offset
})
consumer.subscribe(["delivery-requests"])

try:
    while True:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            print("consumer error:", msg.error())
            continue
        request = json.loads(msg.value())
        print("assigning courier for", request["order_id"])   # your real work here
        consumer.commit(message=msg, asynchronous=False)       # commit AFTER the work
finally:
    consumer.close()   # leave the group cleanly so partitions move immediately

To watch a real group, the CLI shows committed offset, log end and lag per partition. The output below is illustrative of the column layout, not of real numbers.

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

GROUP     TOPIC              PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID
dispatch  delivery-requests  0          100             120             20   c-a-1f3...
dispatch  delivery-requests  1          100             118             18   c-b-77c...

What triggers a rebalance

A rebalance happens when a consumer joins, leaves, or is considered dead. The last case is the one that surprises people. A consumer stays in the group by calling poll regularly and sending background heartbeats. If your processing of one batch takes longer than max.poll.interval.ms, the group coordinator assumes the consumer is stuck, removes it and reassigns its partitions, and the work it was doing is handed to someone else while it is still running. Shrink max.poll.records or speed up processing before raising timeouts, and prefer the cooperative rebalancing assignor so unaffected consumers keep working during a rebalance.

Common mistakes

• Different `group.id` per instance. Each instance becomes its own group and every record is processed once per instance. Symptom: duplicate emails or double dispatches right after scaling out.
• Scaling consumers beyond the partition count. The extra consumers show no assignments and no throughput gain. Symptom: adding pods does not reduce lag.
• Committing before processing. If the process crashes after the commit, the record is never handled. Symptom: a few records missing after every deployment.
• Slow handlers causing endless rebalances. Logs fill with "member was removed from the group" style messages and lag oscillates. The cause is exceeding max.poll.interval.ms, not a Kafka fault.
• Forgetting `flush()` or `close()`. Producers lose queued records on exit; consumers delay failover until the session timeout expires.

How this shows up in real projects and interviews

Interviewers love the scaling question: "your consumers can't keep up, what do you do?". A strong answer checks lag first, then looks at whether the bottleneck is processing or partition count, and notes that adding consumers only helps up to the number of partitions. In production the recurring incidents are rebalance storms from slow processing and offset resets (a group that lost its committed offsets restarting from earliest or latest, depending on auto.offset.reset), so know where that setting is and what it does.