📨 Section 15 · Question #15

What is Rebalancing and When Does It Happen

Rebalancing solves two critical distributed systems challenges:

📨 Kafka 📄 Source Markdown

🟢 Junior Level

30-Second Summary

Rebalancing is a distributed coordination protocol in Apache Kafka during which the broker dynamically reassigns topic partitions among active members of a consumer group (group.id).

Rebalancing solves two critical distributed systems challenges:

  1. Elastic Scalability: Evenly redistributing partition load across consumer instances when scaling out (adding consumers) or scaling in.
  2. High Availability and Fault Tolerance: Automatically detecting failed, crashed, or unresponsive consumer instances, revoking their partition assignments, and delegating them to healthy surviving consumers.

The primary cost of a classic rebalance is a temporary pause in message consumption (Stop-the-World), during which consumers synchronize their state and agree upon the new partition layout.

Real-World Analogy

Imagine a courier delivery brigade:

  • 4 couriers work a shift, each assigned a specific district of the city (a partition).
  • If one courier falls ill and goes home, the central dispatcher (Group Coordinator) halts order dispatches for 1 minute, redraws the district map, and divides the missing courier’s district among the remaining 3 couriers.
  • If a 5th courier arrives on shift, the dispatcher takes a few streets from each existing courier and reassigns them to the new recruit.

When Does Rebalancing Happen?

1. Group Membership Changes:
   ├── A new consumer joins the group (Scale Up)
   ├── A consumer cleanly shuts down (SIGTERM, graceful close, Scale Down)
   └── A consumer crashes unexpectedly (OOM, network partition, kill -9)

2. Consumer Health and Liveness Failures:
   ├── Missing heartbeats beyond session.timeout.ms (hung JVM, long GC pause)
   └── Time between consecutive poll() calls exceeds max.poll.interval.ms

3. Topic Metadata & Topology Changes:
   ├── New partitions are added to an existing topic (via alter topic)
   └── Wildcard subscription (e.g., Pattern "orders-.*") detects a newly created topic

🟡 Middle Level

Protocol Evolution: Eager vs. Cooperative Sticky Rebalancing

Throughout Kafka’s evolution, two fundamentally different rebalancing protocols have been used:

Dimension Eager Rebalancing (Legacy, pre-2.4) Cooperative Sticky Rebalancing (Modern Standard)
Operational Impact Stop-the-World pause across the entire group. Zero downtime for unaffected partitions.
Revocation Strategy All consumers revoke 100% of their partitions upon trigger. Consumers only revoke partitions that actually migrate to other nodes.
Rebalance Rounds Single round of JoinGroup / SyncGroup. Two consecutive non-blocking incremental rounds.
Consumer Group Lag Surges significantly during lengthy pause (5s–40s). Stays flat; uninterrupted processing on surviving partitions.
Implementation RangeAssignor, RoundRobinAssignor CooperativeStickyAssignor

To enable modern cooperative rebalancing in consumer.properties:

partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor

Preserving State: ConsumerRebalanceListener

When a partition is revoked from a consumer, in-flight processing must be finalized and committed immediately. Otherwise, the next owner will reprocess stale messages from the last committed offset:

consumer.subscribe(List.of("orders"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        log.warn("Partitions being revoked: {}. Executing emergency offset commit...", partitions);
        // Synchronously commit offsets before ownership of these partitions is forfeited
        consumer.commitSync();
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        log.info("New partitions assigned: {}. Initializing local state and caches...", partitions);
    }
});

Dual Line of Defense: Heartbeat Thread vs. Main Poll Loop

Kafka clients monitor consumer health via two decoupled, asynchronous mechanisms:

                     Consumer Health Monitoring Architecture
      ┌─────────────────────────────────────────────────────────────────┐
      │                                                                 │
      ▼                                                                 ▼
[ Dedicated Background Heartbeat Thread ]              [ Main Application Poll Thread ]
- Sends heartbeats at heartbeat.interval.ms (3s)       - Executes consumer.poll() and record processing
- Monitored by: session.timeout.ms (45s)               - Monitored by: max.poll.interval.ms (300s / 5m)
- Answers: "Is the JVM alive and connected?"           - Answers: "Is the business logic actively working?"

🔴 Senior Level

Group Coordination Protocol Under the Hood

Rebalancing is orchestrated on the broker by the Group Coordinator (the leader broker of the __consumer_offsets partition to which group.id hashes):

Client 1 (Group Leader)           Group Coordinator                 Client 2 (Member)
       |                                  |                                 |
       |------- 1. FindCoordinator ------>|                                 |
       |<------ Coordinator Metadata -----|                                 |
       |                                  |                                 |
       |------- 2. JoinGroupRequest ----->|<------ JoinGroupRequest --------|
       |        (Proposes Assignors)      |        (Proposes Assignors)     |
       |                                  |                                 |
       |<-- JoinGroupResponse (LEADER) ---|--- JoinGroupResponse (MEMBER) ->|
       |    (Includes full member list)   |                                 |
       |                                  |                                 |
       | 3. Client-Side Assignment Phase  |                                 |
       |    (Leader executes Assignor)    |                                 |
       |                                  |                                 |
       |------- 4. SyncGroupRequest ----->|<------ SyncGroupRequest --------|
       |        (Sends full partition map)|        (Sends empty payload)    |
       |                                  |                                 |
       |<-- SyncGroupResponse (Assignments)<-- SyncGroupResponse (Assignments)|
  1. FindCoordinator: The consumer sends a metadata request to identify which broker hosts the partition of __consumer_offsets mapped to its group.id.
  2. JoinGroup:
    • All group members send a JoinGroupRequest.
    • The coordinator designates one consumer as the Group Leader (typically the first to respond).
    • The coordinator selects the mutually supported partition assignment strategy and returns the full member list exclusively to the Leader.
  3. Partition Assignment:
    • The Group Leader locally computes partition assignments using the chosen strategy (PartitionAssignor).
    • The broker does not calculate assignments in the classic protocol, offloading computational overhead to the client.
  4. SyncGroup:
    • The Leader submits the full assignment plan to the coordinator via SyncGroupRequest.
    • Regular members send an empty SyncGroupRequest and receive their specific assigned partition set in the response.

Static Membership (group.instance.id): Taming Kubernetes Rolling Restarts

In default Kubernetes rolling deployments:

  • Pod 1 restarts $\rightarrow$ Triggers Rebalance 1.
  • Pod 1 joins $\rightarrow$ Triggers Rebalance 2.
  • Across a 20-pod cluster, this causes 40 cascading rebalance cycles, paralyzing processing for up to 30 minutes.

Solution (KIP-345 Static Group Membership):

group.id=order-processing-group
group.instance.id=order-processor-pod-0
session.timeout.ms=120000
  • When group.instance.id is configured, the coordinator identifies the instance as static.
  • When Pod 0 terminates or restarts, the coordinator does not trigger an immediate rebalance.
  • It holds the partition assignments in reserve for session.timeout.ms (e.g., 2 minutes).
  • When the new pod boots with the identical group.instance.id, the coordinator rebinds the existing partition assignments immediately without triggering a group-wide rebalance!

Architectural Evolution: KIP-848 (Next-Gen Server-Side Rebalancing)

Introduced in Apache Kafka 3.7+ and stabilized for Kafka 4.0, the Next-Generation Consumer Protocol (KIP-848) completely overhauls group coordination:

  • Server-Side Assignment: Partition assignment moves from the client-side Group Leader back to the broker-side Group Coordinator.
  • Elimination of Multi-Phase Handshakes: JoinGroup and SyncGroup RPCs are unified into a single, streaming ConsumerGroupHeartbeat RPC.
  • Truly Incremental Transitions: Partitions migrate individually in the background without cluster-wide synchronization barriers.

4 Tricky Questions

1. Why did legacy Eager Rebalancing cause Stop-the-World pauses across the entire group, and how does Cooperative Rebalancing solve this?

Answer:

  • In the Eager protocol, rebalancing operated in a single all-or-nothing step: whenever any trigger occurred, every consumer immediately surrendered ownership of all partitions (revoke). During this window, all processing stopped across the entire consumer group, causing massive consumer lag spikes.
  • In Cooperative Rebalancing (KIP-429), the protocol splits into two non-blocking phases:
    1. Phase 1: Consumers send their current assignments. The coordinator identifies only the differential subset of partitions that must move to achieve balance. Only those specific migrating partitions are revoked; the remaining 90%+ continue processing data uninterrupted.
    2. Phase 2: Migrating partitions are reassigned to their new owners.
  • Downtime is reduced from tens of seconds to sub-second increments, affecting only the partitions undergoing migration.

2. Why can a consumer’s background heartbeat succeed while the broker still evicts it with CommitFailedException?

Answer:

  • Since Kafka 0.10.1, heartbeats are transmitted by an independent daemon thread (HeartbeatThread).
  • As long as this background thread runs, session.timeout.ms will not expire, confirming JVM and network connectivity.
  • However, application progress is strictly governed by the main polling thread via max.poll.interval.ms (default: 300,000 ms / 5 minutes).
  • If the main processing loop fetches a large batch (e.g., 500 records) and blocks on a slow DB query, deadlock, or heavyweight HTTP call for 6 minutes, consumer.poll() is delayed past the deadline.
  • The coordinator detects application stall, evicts the consumer from the group, and triggers a rebalance. When the consumer finally finishes and attempts to commit offsets, the coordinator rejects the request with CommitFailedException.

3. What is a Rebalance Storm cascade, and how do you remediate it in production?

Answer:

  • Anatomy of the Cascade:
    1. A traffic spike causes consumer $C_1$ to poll a full batch of complex records.
    2. Processing exceeds max.poll.interval.ms. The coordinator evicts $C_1$ and redistributes its partitions to $C_2$ and $C_3$.
    3. $C_2$ and $C_3$ now inherit uncommitted partitions on top of their own load. They poll the heavy records, breach max.poll.interval.ms, and are evicted as well.
    4. The entire consumer group enters an infinite loop of continuous rebalancing with zero forward progress.
  • Production Remediation:
    • Reduce max.poll.records (e.g., from 500 to 50) so each batch can be guaranteed to finish well within the interval.
    • Increase max.poll.interval.ms to provide sufficient headroom for worst-case downstream latency.
    • Adopt CooperativeStickyAssignor to avoid cluster-wide stalls.
    • Implement per-record processing timeouts and offload asynchronous work to worker pools with circuit breakers.

4. Why is it dangerous to hold mutable, unpersisted state in consumer memory without implementing ConsumerRebalanceListener?

Answer:

  • If a consumer maintains in-memory aggregations, windowed state, or user sessions (e.g., inside a local ConcurrentHashMap) and a rebalance occurs:
    1. Partitions are revoked and assigned to a different consumer instance that lacks this in-memory context.
    2. If the departing consumer does not flush buffered state to persistent storage (e.g., DB, Redis) and execute commitSync() within onPartitionsRevoked(), two critical failure modes occur:
      • The new consumer re-reads uncommitted records, causing duplicate state mutation.
      • Any uncommitted aggregated calculations held only in the previous consumer’s heap are permanently lost.
  • For robust stateful stream processing, applications should rely on Kafka Streams with local RocksDB backing and changelog topics rather than raw KafkaConsumer memory structures.

🎯 Interview Cheat Sheet

Core Concepts

  • Definition: Dynamic reassignment of topic partitions among active members of a consumer group.
  • Core Triggers: Consumer joins, clean leaves, session.timeout.ms expiration (network/crash), or max.poll.interval.ms expiration (processing hang).
  • Eager vs. Cooperative: Always configure CooperativeStickyAssignor in modern production systems to eliminate Stop-the-World group pauses.
  • Static Membership: Use group.instance.id (KIP-345) to prevent rebalance storms during Kubernetes pod rolling restarts.
  • ConsumerRebalanceListener: Essential for committing in-flight offsets and flushing local state inside onPartitionsRevoked().

Key Configuration Reference

| Parameter | Operating Thread | Purpose | Default | Failure Symptom | | :— | :— | :— | :— | :— | | session.timeout.ms | Heartbeat thread | Broker detection of dead nodes or network failure | 45,000 ms (45s) | Node evicted; partitions reassigned | | heartbeat.interval.ms| Heartbeat thread | Frequency of heartbeat pings sent to coordinator | 3,000 ms (3s) | Spurious rebalances if $\ge \frac{1}{3} \times$ session timeout | | max.poll.interval.ms | Main poll thread | Maximum allowed elapsed time between poll() calls | 300,000 ms (5m) | CommitFailedException; rebalance storm | | max.poll.records | Main poll thread | Upper bound on records returned in a single poll() | 500 records | Batch processing breaches max.poll.interval.ms |

Red Flags to Avoid

  • ❌ “Rebalancing physically migrates partition data files between Kafka brokers.” (Rebalance only reassigns client-side partition ownership).
  • ❌ “Heartbeats are sent synchronously during consumer.poll().” (Heartbeats run asynchronously on a dedicated background HeartbeatThread).
  • ❌ “The Kafka broker computes the partition assignments in standard rebalancing.” (In the classic protocol, the Group Leader consumer computes assignments and sends them to the broker).
  • ❌ “Rebalancing always blocks all message processing across the group.” (Modern CooperativeStickyAssignor executes incremental reassignment without interrupting unaffected partitions).