Rebalance Process #

In Apache Kafka’s Consumer Group architecture, the partition reassignment process, or Rebalance Process, is a crucial mechanism guaranteeing high availability and dynamic horizontal scalability. Automatic rebalance is triggered every time a new consumer joins, an old consumer intentionally leaves (graceful shutdown), or when a consumer crashes suddenly. However, although rebalance is a very sophisticated safety feature, it has a significant performance impact on our system throughput. Understanding the internal lifecycle of the rebalance protocol, how to detect its triggers, and how to implement custom listeners to secure our data before partitions are moved to other consumers is a mandatory skill for distributed system engineers.

Why and When Does Rebalance Happen? #

The rebalance process isn’t something that happens randomly. It’s a direct response from the Kafka cluster (specifically the Group Coordinator) to changes in membership topology or topic metadata.

Here are the five main triggers of the rebalance process in production:

  1. New Consumer Joins (Scale-Up): When we add a new application pod to help speed up data processing, that new instance sends a join request with the same group.id, triggering a rebalance to divide existing partitions fairly.
  2. Consumer Leaves Planned (Graceful Shutdown / Scale-Down): When we do application updates (rolling upgrade) or reduce server capacity, application containers close. Before dying, the consumer sends a LeaveGroup signal to the Group Coordinator, triggering a rebalance so the abandoned partitions are immediately transferred to remaining consumers.
  3. Missing Heartbeat (Unplanned Crash / Network Partition): If an application instance crashes from a power outage, OutOfMemory, or network isolation, it stops sending heartbeats. After the session.timeout.ms limit passes, the Group Coordinator declares that consumer dead and starts a rebalance.
  4. Livelock (Main Thread Stuck): If our application’s main thread spends too long processing business logic (for example, processing a heavy database batch query) past the max.poll.interval.ms limit, the consumer considers itself problematic and sends an exit instruction to the broker, triggering a rebalance even though the server is physically still on.
  5. Topic Metadata Changes: If a Kafka administrator adds new partitions to a topic we’re consuming (for example, doing partition expansion from 10 to 20 partitions), a rebalance happens so consumers can distribute those new partitions.

Internal Rebalance Protocol Lifecycle #

To understand how data stops during rebalance, we must look at the difference between the two main rebalance protocols: Eager Rebalance (old protocol) and Cooperative Rebalance (new protocol).

1. Eager Rebalance (Classic Protocol) #

This protocol applies the “release everything, search again” principle. Here’s its phase-by-phase journey:

sequenceDiagram
    participant C1 as Consumer 1 (Leader)
    participant C2 as Consumer 2
    participant GC as Group Coordinator (Broker)

    Note over C1,C2: Phase 1: Trigger Detection
    C2->>GC: LeaveGroup Request (C2 is shut down)
    GC->>C1: Detect metadata change during Heartbeat / Poll

    Note over C1: Phase 2: Partition Revocation
    C1->>C1: Stop data processing, release all partitions (Stop-The-World)
    
    Note over C1: Phase 3: Sending JoinGroup Request
    C1->>GC: JoinGroup Request
    GC-->>C1: JoinGroup Response (C1 designated as Group Leader)

    Note over C1: Phase 4: Partition Assignment Calculation
    C1->>C1: Recalculate partition allocation for all remaining members

    Note over C1: Phase 5: Synchronization (SyncGroup)
    C1->>GC: SyncGroup Request (Sending the new allocation calculation result)
    GC-->>C1: SyncGroup Response (Receiving the new allocation)

During the process from Phase 2 to Phase 5 completion, all consumers can’t process new data. The processing phase stops completely (Stop-The-World). If the cluster has hundreds of partitions and dozens of consumers, this process can take up to tens of seconds, triggering queue buildup (lag spike).

2. Cooperative Rebalance (Incremental) #

This modern protocol (introduced since Kafka 2.4) operates cooperatively and incrementally. Instead of stripping all partitions from all consumers simultaneously, this protocol divides the move into two small rebalance cycles without disrupting processing:

  1. First Cycle: When there’s a membership change (for example, new consumer C3 joins), the Group Coordinator is contacted. The client calculates the new allocation and realizes partition P2, previously held by C2, must move to C3.
  2. Partial Revocation: The consumer client C2 releases only partition P2. In this phase, consumer C1 holding P0 and P1 is not affected at all and keeps processing data without stopping.
  3. Second Cycle: After P2 is freed, a very fast second rebalance executes only to assign P2 to the new consumer C3. This cycle happens very quickly in milliseconds because there’s no global partition release.

Consumer Group State Machine on the Broker Coordinator Side #

Inside the broker acting as the Group Coordinator, there’s an internal state machine managing group membership transitions. Understanding these transitions makes it easier to read Kafka broker log files (server.log) when debugging.

Here are the five group membership states in Kafka:

  • Empty: The group has no active members, but the last committed offset metadata for that group is still stored in the internal __consumer_offsets topic. This state happens if all application pods are shut down.
  • PreparingRebalance: The Group Coordinator has received a trigger (like a lost heartbeat or a new join request) and is preparing to redistribute partitions. The broker is waiting for all active consumers to send JoinGroup requests before the deadline.
  • CompletingRebalance: All active members have sent JoinGroup requests. The Coordinator designates the Group Leader and is waiting for that leader to return the partition division calculation result via a SyncGroup request.
  • Stable: The division transaction is complete. All active consumers have received their partition allocations and are actively reading data and sending periodic heartbeats. This is the ideal production system state.
  • Dead: The group is considered permanently dead because of long inactivity, or the group is being manually deleted by system administrators via CLI commands.

The Danger of Rebalance Storms #

One of the most dangerous phenomena in large-scale production environments is the Rebalance Storm. This condition happens when a consumer group enters an endless rebalance cycle that cripples the system.

Storm Trigger: Garbage Collection Pauses (GC Pauses) #

In Java Virtual Machine (JVM)-based applications like Java, Scala, or Kotlin, automatic memory management uses the Garbage Collector. If our application experiences a large buildup of garbage objects in heap memory, the JVM triggers a full cleanup cycle known as Stop-The-World Garbage Collection (Major GC).

  • During the Major GC process, all JVM threads (including the Kafka client’s background heartbeat thread) are totally frozen by the CPU.
  • If this GC pause lasts 50 seconds, while our session.timeout.ms configuration is set to 45 seconds, the coordinator broker concludes our consumer died because it didn’t receive a heartbeat within 45 seconds.
  • The broker triggers a rebalance process to move the partitions to other consumer pods.
  • When the GC pause finishes, the first consumer realizes it was forcibly removed from the group. It immediately sends a rejoin request (JoinGroup).
  • This rejoin request triggers yet another rebalance.
  • During the new rebalance, the workload moves to another pod which can trigger a memory spike in that pod, triggering a Major GC in the second pod, and this disaster cycle repeats continuously spreading across all consumer pods.

Rebalance Storm Prevention Recommendations #

To secure our application from Rebalance Storm threats, we’re advised to apply the following mitigation steps:

  1. Use G1GC or ZGC: Configure the JVM to use a modern Garbage Collector designed to minimize thread freeze pauses (for example, using the -XX:+UseG1GC or -XX:+UseZGC options).
  2. Increase Session Timeout: Don’t set session.timeout.ms too close to the heartbeat interval. In dense production environments, 45000 (45 seconds) to 60000 (60 seconds) are safe choices.
  3. Optimize Heap Memory: Do regular memory profiling to avoid memory leaks that trigger consecutive GCs.

Rebalance Listener: Saving State Before Rebalance #

One of the most fatal mistakes when managing offsets manually or storing local state in application memory is not responding to rebalance events correctly. When Kafka revokes partition access rights from our application to hand them to another consumer, we must ensure that:

  1. All messages being processed finish executing orderly.
  2. The last offset commit is successfully sent to the broker before that partition is held by someone else (preventing duplication).
  3. Local memory caches related to those partitions are safely cleaned.

To handle this need, the Kafka SDK provides the ConsumerRebalanceListener interface. This listener has two main callback functions:

1. onPartitionsRevoked(Collection<TopicPartition> partitions) #

  • When called: Right before partitions are officially revoked from the current consumer.
  • Purpose: This is where we must do a synchronous offset commit (commitSync()) for the last data we finished processing. Because these partitions will soon be consumed by another node, committing offsets here is our last line of defense against data duplication disasters.

2. onPartitionsAssigned(Collection<TopicPartition> partitions) #

  • When called: Right after new partitions are allocated to the current consumer (before data reading via .poll() resumes).
  • Purpose: Here we can initialize local state, look up custom offset positions from an external database using consumer.seek(), or record monitoring metric logs.

Implementation Code: Securing Offsets via a Rebalance Listener #

Let’s look at the code implementation difference between manual consumer writing without rebalance handling (which triggers data leaks) versus a robust custom Listener implementation.

Java SDK Anti-Pattern: Ignoring Rebalance Events on Manual Commit #

If we disable auto-commit but don’t register a Rebalance Listener, every time a rebalance happens (for example, when redeploying the application), messages being processed in memory get re-consumed by the new consumer because the last offset wasn’t committed to the broker in time.

// ANTI-PATTERN: Consuming data with manual commit without handling rebalance
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-worker");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); // Manual commit active
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// ✗ DON'T: Only subscribe to the topic without attaching a Rebalance Listener.
// If a rebalance happens mid-batch processing, the last commit offset
// is lost and triggers massive duplicate processing on the new pod.
consumer.subscribe(Collections.singletonList("order-events"));

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            processOrder(record.value());
        }
        // Commit happens at the end of the batch loop
        consumer.commitSync();
    }
} finally {
    consumer.close();
}

Java SDK Solution: Implementing ConsumerRebalanceListener Orderly #

By implementing ConsumerRebalanceListener, we capture the moment right before partitions are moved to force a synchronous commit of the last offsets, as well as accurately track back the read position of new partitions.

// CORRECT: Using ConsumerRebalanceListener for safe offset synchronization
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-worker");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// Local map to track the last uncommitted offsets in memory
final Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();

ConsumerRebalanceListener rebalanceListener = new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        log.info("Rebalance Process Detected! Partitions will be revoked: {}", partitions);
        
        // ✓ CORRECT: Immediately synchronously commit the last offsets being processed in memory.
        // This guarantees the new consumer taking over these partitions continues
        // from the correct offset, avoiding duplicate data processing.
        try {
            consumer.commitSync(currentOffsets);
        } catch (Exception e) {
            log.error("Failed to do an emergency offset commit during rebalance", e);
        }
        
        // Clear the offset tracking for revoked partitions
        currentOffsets.keySet().removeAll(partitions);
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        log.info("New partitions successfully assigned to this consumer: {}", partitions);
        
        // Here we can initialize resources or do seek() if needed.
        // Example: Ensuring the read pointer starts right after the committed offset on the broker.
        for (TopicPartition partition : partitions) {
            OffsetAndMetadata committed = consumer.committed(partition);
            if (committed != null) {
                log.info("Starting partition {} from committed offset {}", partition, committed.offset());
                consumer.seek(partition, committed.offset());
            }
        }
    }
};

// ✓ Register subscribe with the custom listener
consumer.subscribe(Collections.singletonList("order-events"), rebalanceListener);

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            processOrder(record.value());
            
            // Track the next record offset to be consumed (+1)
            currentOffsets.put(
                new TopicPartition(record.topic(), record.partition()),
                new OffsetAndMetadata(record.offset() + 1)
            );
        }
        
        if (!records.isEmpty()) {
            consumer.commitAsync(currentOffsets, null);
        }
    }
} catch (Exception e) {
    log.error("A fatal error occurred in the consumer loop", e);
} finally {
    try {
        // Final emergency commit before closing the connection
        consumer.commitSync(currentOffsets);
    } finally {
        consumer.close();
    }
}

Summary #

  • Rebalance Trigger — The partition reassignment event triggered by new consumer additions, sudden old consumer deaths, or topic partition metadata changes.
  • Stop-The-World Effect — In the classic Eager Rebalance protocol, all data processing is globally paused temporarily while all partitions are revoked and reallocated from scratch.
  • Cooperative Rebalance — The modern (incremental) protocol minimizing processing interruptions by only revoking partitions changing ownership without resetting other stable consumer states.
  • ConsumerRebalanceListener — The mandatory interceptor interface allowing applications to capture partition revocation and allocation events in real-time.
  • onPartitionsRevoked — The crucial callback for immediately doing a synchronous offset commit (commitSync()) right before partition ownership changes hands to prevent data duplication risk.
  • onPartitionsAssigned — The post-rebalance callback used for state initialization, cache cleanup, or jumping read offset positions (seek) from external database storage.
  • Rebalance Storm Mitigations — Long JVM GC pauses can trigger wild rebalance storms, which can be overcome by raising the session timeout or switching to the ZGC/G1GC engine.
  • Graceful Shutdown — Always call the .close() function in the application’s finally block so the coordinator immediately triggers a rebalance without waiting for the session timeout to end.

← Previous: Partition Assignment Strategy Next: At-Least-Once vs At-Most-Once →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact