sdlc.dev

Kafka path · Writing and reading · Topic 4 of 12

Consumer groups, rebalancing and offsets

Not started

Consumers that share a group.id form a consumer group. The group's coordinator, a broker, divides the topic's partitions among the members so each partition has exactly one owner, and stores each partition's committed offset in the internal __consumer_offsets topic. When a member arrives, leaves or dies, partitions are reassigned: a rebalance. The trace below follows one partition through a commit, a crash and a handover.

Why it matters: Almost every duplicate, every gap and every mysterious pause in a Kafka consumer comes from when offsets are committed and what happens during a rebalance.

  1. 1A group shares partitions, not records

    Each partition is assigned to exactly one member of the group. More members than partitions leaves some idle.

  2. 2Committed offset means next to read

    The group stores, per partition, the offset of the next record to process. A new owner starts there.

  3. 3Commit after processing gives at-least-once

    If a consumer crashes after processing but before committing, the next owner processes those records again.

  4. 4Rebalances move partitions

    Members joining, leaving, crashing or missing max.poll.interval.ms trigger a rebalance. Kafka 4.0's new protocol makes it incremental and server-driven.

1

Watch it run

Play it or step through. Every step says what just happened.

const log = ["o1", "o2", "o3", "o4", "o5"];let committed = 0; // kept by the group coordinatorconst poll = (from, max) =>  log.slice(from, from + max).map((v, i) => ({ offset: from + i, v })); // Consumer A owns the partition: process a batch, then commitlet batch = poll(committed, 3);for (const r of batch) console.log("A processed", r.offset, r.v);committed = batch[batch.length - 1].offset + 1; // A processes the next batch but crashes before committingbatch = poll(committed, 3);for (const r of batch) console.log("A processed", r.offset, r.v); // Rebalance: B gets the partition and resumes from the commitbatch = poll(committed, 3);for (const r of batch) console.log("B processed", r.offset, r.v);committed = batch[batch.length - 1].offset + 1;console.log("committed", committed);

Partition 0

The log; reading never removes records

empty

Consumer A

Records it has fetched

empty

Consumer B

Records it has fetched

empty

Committed offset

Next offset to read, stored by the group

empty

Console0 lines

    Step 0 of 12

    Nothing has run yet. One partition, a group with consumers A and B, and A owns the partition.

    A model of one partition, not the Kafka client API. In Java the commit is consumer.commitSync() after the processing loop.

    2

    Try it yourself

    Change the code, predict the order, then run it for real in your browser.

    ⌘/Ctrl + Enter
    OutputNot run yet

    Press Run to see what your code prints, in order.

    Runs in this browser, in an isolated frame with no access to this page. console.log, info, warn and error are shown; code runs as a classic script, so use an async function for await.

    3

    Check yourself

    3 questions. Get them all right to finish this topic.

    1. 1.A consumer processes offsets 3 and 4, then crashes before committing. The last commit was 3. Where does the next owner start?
    2. 2.With enable.auto.commit=true (the default), when are offsets committed?
    3. 3.A topic has 6 partitions and the group has 8 consumers. How many consumers receive records?

    0 of 3 right so far. Get all three to finish this topic.

    On a real cluster

    Two console consumers in group shipping on a three-partition topic, using the new protocol (group.protocol=consumer):

    bin/kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --describe --group shipping --members --verbose
    GROUP     CONSUMER-ID             CLIENT-ID  #PARTITIONS  CURRENT-EPOCH  CURRENT-ASSIGNMENT  ...shipping  CNZp_LDPRGCdXHKzF638Mw  worker-1   2            2              orders:0,1shipping  WUzGEJhgTdyzZpJbweOUeg  worker-2   1            2              orders:2

    The group's state showed the server-side assignor uniform. Describing a group without --members shows lag per partition, the gap between the committed offset and the end of the log:

    GROUP    TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAGbilling  orders  0          3               4               1billing  orders  1          1               4               3

    --reset-offsets --to-earliest (or --to-datetime, --shift-by) moves a stopped group's position, which is how you replay.

    Settings that cause trouble

    • enable.auto.commit=true (default) commits every auto.commit.interval.ms (5 s) whatever poll() has returned, done or not.
    • max.poll.interval.ms (5 minutes): a member that does not call poll() within it is considered dead and its partitions are moved. Slow processing per batch looks like a crash. Lower max.poll.records (default 500) or speed up the handler.
    • auto.offset.reset decides where a group with no committed offset starts: latest by default, so a new group skips history unless you set earliest.
    • Static membership (group.instance.id) lets a restarting member keep its partitions within the session timeout instead of triggering two rebalances.

    Kafka 4.0 made the new rebalance protocol (KIP-848) generally available. The broker computes assignments and moves partitions incrementally, without stopping the whole group. Clients still default to group.protocol=classic; you opt in with group.protocol=consumer.

    Go deeper