What Happens When You Add a New Consumer to the Group

When a new consumer instance connects to an active Consumer Group in Apache Kafka, the cluster initiates a rebalance: 4. Resumption: The new consumer begins reading its newly as...

🟢 Junior Level

30-Second Summary

When a new consumer instance connects to an active Consumer Group in Apache Kafka, the cluster initiates a rebalance:

Registration: The new instance sends a registration request to the broker acting as the Group Coordinator.
Rebalance Trigger: The coordinator notifies all existing group members that group membership has changed.
Partition Revocation: A portion of partitions is revoked from existing consumers and assigned to the newcomer to ensure equal workload distribution.
Resumption: The new consumer begins reading its newly assigned partitions starting from the latest committed offsets stored in the internal __consumer_offsets topic.

Real-World Analogy

Think of a team of postal couriers delivering packages across city districts:

A city consists of 6 delivery districts (6 partitions).
2 couriers are currently working: Courier A covers Districts 1, 2, and 3; Courier B covers Districts 4, 5, and 6.
A 3rd courier (Courier C) arrives at the depot to help. The dispatcher calls a brief meeting:
- Courier A hands over District 3 to the new courier.
- Courier B hands over District 6 to the new courier.
Now each courier handles exactly 2 districts. The workload is evenly divided, delivery latency drops, and overall throughput increases.

Partition Redistribution Topology

Before: 2 Consumers on 4 Partitions
  Partition 0 ──┐
  Partition 1 ──┴──► [ Consumer 1 ]
  Partition 2 ──┐
  Partition 3 ──┴──► [ Consumer 2 ]

After Consumer 3 Joins: [ REBALANCE ]
  Partition 0 ──┐
  Partition 1 ──┴──► [ Consumer 1 ]
  Partition 2 ──────► [ Consumer 2 ]
  Partition 3 ──────► [ Consumer 3 ]  <-- Newcomer claims Partition 3

🟡 Middle Level

The Step-by-Step Join Protocol

The join sequence coordinates membership changes safely across distributed nodes:

[ New Consumer C3 ]              [ Group Coordinator (Broker) ]          [ Existing Consumers C1 & C2 ]
        │                                      │                                      │
        │ 1. JoinGroup (memberId="")           │                                      │
        ├─────────────────────────────────────►│                                      │
        │ 2. Error: MEMBER_ID_REQUIRED         │                                      │
        │◄─────────────────────────────────────┤                                      │
        │                                      │                                      │
        │ 3. JoinGroup (with generated ID)     │                                      │
        ├─────────────────────────────────────►│                                      │
        │                                      │ 4. Coordinator waits for C1 & C2     │
        │                                      │    to invoke poll() & JoinGroup      │
        │                                      │◄─────────────────────────────────────┤
        │                                      │    (within max.poll.interval.ms)     │
        │ 5. Elects Leader (typically C1)      │                                      │
        │ 6. Leader runs Assignor.assign()     │                                      │
        │ 7. SyncGroup Request (sends plan)    │ 7. SyncGroup Request                 │
        ├─────────────────────────────────────►│◄─────────────────────────────────────┤
        │ 8. SyncGroup Response (P3 assigned)  │ 8. SyncGroup Response (P3 revoked)   │
        │◄─────────────────────────────────────┼─────────────────────────────────────►│

Safe State Commit via `ConsumerRebalanceListener`

When the coordinator revokes a partition from an active consumer to transfer it to a newcomer, the losing consumer must execute clean-up logic inside onPartitionsRevoked():

consumer.subscribe(List.of("orders"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        log.info("Partitions being revoked: {}. Flusing state...", partitions);
        // 1. Finish processing in-flight records
        // 2. Synchronously commit offsets so newcomer does not read duplicate records!
        consumer.commitSync();
        // 3. Flush downstream transactional database buffers
        flushDatabaseBuffers(partitions);
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        log.info("Newly assigned partitions: {}", partitions);
        // Initialize local caches or seek offsets if necessary
    }
});

🔴 Senior Level

Why `CooperativeStickyAssignor` Requires Two Rebalance Rounds

Under legacy Eager Rebalancing, all consumers surrendered 100% of their partitions in a single round, causing a disruptive group-wide Stop-the-World pause (5–30+ seconds). In CooperativeStickyAssignor (KIP-429), rebalancing is broken into two lightweight, non-blocking rounds so unimpacted partitions continue processing uninterrupted:

ROUND 1: Identify and Revoke Migrating Partitions
  - C3 sends JoinGroup. C1 and C2 continue processing records uninterrupted!
  - Group Leader evaluates current assignments: C1 owns {P0, P1}, C2 owns {P2, P3}.
  - Leader determines: P3 must move to newcomer C3.
  - Round 1 outcome: C1 retains {P0, P1}, C2 revokes {P3} (retains {P2}), C3 receives empty set {}.
  - C2 triggers onPartitionsRevoked({P3}), commits offsets, and surrenders P3.
  - C1 and C2 (on P2) never pause consumption for even a millisecond!

ROUND 2: Assign Released Partition to Newcomer
  - Coordinator confirms that P3 is officially unassigned.
  - A brief second sync completes:
  - C3 receives {P3}, C1 retains {P0, P1}, C2 retains {P2}.
  - Total group rebalance finishes with zero global downtime!

Zombie Fencing and Member Epochs

What happens if consumer C2 pauses during a massive Garbage Collection pause, causing the coordinator to assume it is dead, revoke partition P3, and reassign P3 to newcomer C3? When C2 revives, both consumers could concurrently process P3 (Split-Brain hazard).

The Architectural Protection — Generation ID / Member Epoch:

Every time group membership changes, the coordinator increments the group Generation ID (or Member Epoch in KIP-848).
Every consumer request (OffsetCommitRequest, FetchRequest) must carry its assigned epoch.
When the stalled consumer C2 wakes up and attempts to commit offsets, the broker checks its epoch, detects that it is obsolete, and rejects the commit with CommitFailedException or FencedMemberEpochException.

4 Tricky Questions

1. Why does a new consumer send a `JoinGroup` request TWICE when first connecting to a Kafka cluster?

Answer:

In early Kafka versions, clients generated their own member.id or the coordinator assigned one immediately upon rebalance initiation. This caused phantom rebalances whenever unstable clients sent transient requests.
Since KIP-394, Kafka enforces a strict two-phase handshake:
1. The new consumer sends its first JoinGroupRequest with an empty string member.id = "".
2. The broker does not initiate a rebalance. Instead, it generates a unique persistent identifier (e.g., client-id-uuid), reserves it, and responds with an error code: MEMBER_ID_REQUIRED.
3. The consumer extracts the assigned member.id and immediately sends a second JoinGroupRequest carrying this verified ID.
The coordinator initiates the rebalance only upon receiving the second request, safeguarding the cluster against accidental rebalances triggered by rogue or dying network connections.

2. What happens if an active consumer is in the middle of a 15-second database transaction when a new consumer joins, assuming `max.poll.interval.ms = 30` seconds?

Answer:

The broker cannot preemptively interrupt or cancel Java code executing inside a consumer’s message handler. It simply sets a RebalanceInProgress flag in background heartbeat responses.
The Group Coordinator patiently waits for the active consumer to finish its 15-second transaction and invoke its next consumer.poll() call.
Once poll() is called, the Kafka client detects the rebalance flag, invokes onPartitionsRevoked(), flushes state, and submits its JoinGroupRequest.
Consequently, the rebalance start is delayed by 15 seconds. All other consumers in the group (including the newly connected instance) will wait in the JoinGroup barrier phase until this active consumer completes its work.

3. A new consumer joins, rebalance finishes, and it is assigned Partition 2. From which exact offset will it start consuming messages?

Answer:

The consumer determines its starting offset through a deterministic hierarchy:
1. The consumer issues an OffsetFetchRequest to the coordinator for Partition 2.
2. The coordinator reads the latest committed offset for that specific group and partition from the __consumer_offsets system topic.
3. If a committed offset is found (e.g., offset 45100), the new consumer begins reading strictly from 45100.
4. If no committed offset exists (e.g., first-time consumption or offset expired via offsets.retention.minutes), the consumer consults its local auto.offset.reset property:
  - earliest: reads from the oldest record available on disk (offset 0 or retention floor).
  - latest: seeks to the end of the partition log, reading only newly published records.
  - none: throws a fatal NoOffsetForPartitionException.

4. What happens if a new consumer joins an active group, but total consumer count now exceeds the topic’s partition count (e.g., a 5th consumer on a 4-partition topic)?

Answer:

The rebalance executes successfully without throwing exceptions.
The configured PartitionAssignor allocates the 4 existing partitions among 4 of the group members.
The 5th consumer receives an empty partition assignment: assignedPartitions = [].
It enters the Idle state:
- It continues sending periodic HeartbeatRequest RPCs every heartbeat.interval.ms to keep its session alive.
- Calls to consumer.poll() immediately return an empty record batch (ConsumerRecords.empty()).
- It performs no message processing and acts as an immediate Hot Standby, ready to inherit a partition if an active consumer crashes.

🎯 Interview Cheat Sheet

Core Concepts at a Glance

Trigger Effect: Adding a consumer triggers a Rebalance to redistribute partitions.
Two-Phase Handshake (KIP-394): First JoinGroup receives a verified member.id; the second triggers rebalance.
Legacy Eager Protocol: Causes a Stop-the-World group pause (5–30+ seconds).
Cooperative Sticky (KIP-429): Executes in 2 incremental phases without stopping unaffected partitions.
ConsumerRebalanceListener: Mandatory hook to invoke commitSync() inside onPartitionsRevoked().
Starting Offset: Fetched from __consumer_offsets; falls back to auto.offset.reset if absent.
Zombie Protection: Generation ID / Member Epoch blocks obsolete consumer commits with CommitFailedException.

Rebalance Protocol Comparison on Scale-Up

Metric	Eager Protocol	Cooperative Sticky Protocol
Active Ingestion	Halts across 100% of consumer instances	Continues on all unimpacted partitions
Rebalance Rounds	1 monolithic round with global freeze	2 lightweight non-blocking rounds
Latency Impact	Severe lag spike across entire topic	Imperceptible latency bump on migrated partition only

Red Flags to Avoid

❌ “A new consumer immediately reads messages in parallel with the old consumer on the same partition.” (Partitions are mutually exclusive; reading starts only after the old owner commits and releases).
❌ “A new consumer always reads from offset 0.” (It resumes from the last committed offset in __consumer_offsets).
❌ “A rebalance forcefully cancels active Java worker threads mid-execution.” (The coordinator waits for the current poll() cycle to conclude within max.poll.interval.ms).
❌ “Adding a consumer beyond partition count throws a KafkaException.” (Extra consumers join normally and sit idle).

What is Rebalancing and When Does It Happen — Rebalancing lifecycle and triggers
How Does Consumer Balancing Work in a Group — Assignment assignors
What is Consumer Group — Group architecture
Can You Have More Consumers Than Partitions — Idle standby instances

🟢 Junior Level

30-Second Summary

Real-World Analogy

Partition Redistribution Topology

🟡 Middle Level

The Step-by-Step Join Protocol

Safe State Commit via ConsumerRebalanceListener

🔴 Senior Level

Why CooperativeStickyAssignor Requires Two Rebalance Rounds

Zombie Fencing and Member Epochs

4 Tricky Questions

1. Why does a new consumer send a JoinGroup request TWICE when first connecting to a Kafka cluster?

2. What happens if an active consumer is in the middle of a 15-second database transaction when a new consumer joins, assuming max.poll.interval.ms = 30 seconds?

3. A new consumer joins, rebalance finishes, and it is assigned Partition 2. From which exact offset will it start consuming messages?

4. What happens if a new consumer joins an active group, but total consumer count now exceeds the topic’s partition count (e.g., a 5th consumer on a 4-partition topic)?

🎯 Interview Cheat Sheet

Core Concepts at a Glance

Rebalance Protocol Comparison on Scale-Up

Red Flags to Avoid

Related Topics

Safe State Commit via `ConsumerRebalanceListener`

Why `CooperativeStickyAssignor` Requires Two Rebalance Rounds

1. Why does a new consumer send a `JoinGroup` request TWICE when first connecting to a Kafka cluster?

2. What happens if an active consumer is in the middle of a 15-second database transaction when a new consumer joins, assuming `max.poll.interval.ms = 30` seconds?