How Kafka Consumer 4.0’s New Rebalance Protocol Eliminates the Two-Phase Bottleneck

// Classic Protocol Flow - First Rebalance
Consumer A: owns partitions [1, 3]
Consumer B: owns partitions [2]
Consumer C: joining group

// Phase 1: All consumers must rejoin
1. Consumer C sends JoinGroup request
2. Consumers A & B discover rebalance via heartbeat response
3. All consumers send JoinGroup requests with current assignments
4. Coordinator waits for ALL members before proceeding
5. Leader (Consumer A) computes new assignment
6. Assignment removes partition 3 from A (unassigned state)

// Phase 2: Second rebalance required
7. Consumer A completes revocation, triggers new rebalance
8. All consumers rejoin again
9. Leader reassigns partition 3 to Consumer C

This two-phase approach creates multiple points of failure. If any consumer becomes slow during either phase, the entire group waits. The OSO team has observed rebalances taking over 30 seconds in groups with just 10 members when one instance experienced garbage collection pauses.

# Classic Consumer Configuration - Prone to Synchronisation Issues
group.protocol=classic
partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor
session.timeout.ms=30000
heartbeat.interval.ms=3000
max.poll.interval.ms=300000

# These client-side settings create coordination complexity
enable.auto.commit=true
auto.commit.interval.ms=5000

During rebalancing with this configuration, consumers cannot commit offsets, meaning any processing progress since the last commit is lost if failures occur. This represents both a performance and reliability concern that scales poorly.

// New Protocol Flow - Single Incremental Rebalance
Consumer A: owns partitions [1, 3]
Consumer B: owns partitions [2]  
Consumer C: joining group

// Single-phase incremental assignment
1. Consumer C starts heartbeating to join
2. Coordinator computes new assignment (move partition 3: A → C)
3. Consumer A's next heartbeat receives revocation instruction
4. Consumer A revokes partition 3, acknowledges via heartbeat
5. Consumer C's next heartbeat receives partition 3 assignment
6. Rebalance complete - no group-wide synchronisation required

The configuration for this new approach is remarkably simpler:

properties# New Consumer Configuration - Server-Side Intelligence
group.protocol=consumer
group.remote.assignor=uniform

# No client-side assignment strategy needed
# No client-side session management
# Session and heartbeat managed by broker per-group

// Demonstrating the difference in consumer behaviour
public class ConsumerRebalanceComparison {
    
    // Classic Protocol - All consumers stop during rebalance
    public void classicRebalance() {
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            
            // During rebalance: NO records returned, NO commits allowed
            // All consumers wait for slowest member
            
            for (ConsumerRecord<String, String> record : records) {
                processRecord(record);
            }
            
            // Commits fail during rebalance
            consumer.commitSync(); // May throw CommitFailedException
        }
    }
    
    // New Protocol - Unaffected consumers continue processing
    public void newProtocolRebalance() {
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            
            // During rebalance: Unaffected partitions continue processing
            // Only consumers losing/gaining partitions pause briefly
            
            for (ConsumerRecord<String, String> record : records) {
                processRecord(record);
            }
            
            // Commits succeed during rebalance for stable partitions
            consumer.commitSync(); // Success for unaffected partitions
        }
    }
}

public class RebalanceAwareConsumer {
    
    public void demonstrateOffsetCommitDuringRebalance() {
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(getNewProtocolConfig());
        
        consumer.subscribe(Arrays.asList("orders", "payments"), new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                // With new protocol: This only affects partitions being moved
                System.out.println("Revoking partitions: " + partitions);
                
                // Commit is allowed and succeeds for stable partitions
                consumer.commitSync();
            }
            
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                // Surgical assignment - only new partitions
                System.out.println("Assigned new partitions: " + partitions);
            }
        });
        
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            
            for (ConsumerRecord<String, String> record : records) {
                processRecord(record);
                
                // This commit succeeds even during rebalancing
                // for partitions not involved in the migration
                if (shouldCommit()) {
                    consumer.commitSync(); // No CommitFailedException for stable partitions
                }
            }
        }
    }
    
    private Properties getNewProtocolConfig() {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "orders-processing-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        
        // Enable new protocol
        props.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, "consumer");
        props.put("group.remote.assignor", "uniform");
        
        return props;
    }
}

// Performance Comparison Test
public class RebalancePerformanceTest {
    
    public void benchmarkRebalanceTime(String protocol, int consumerCount) {
        List<KafkaConsumer<String, String>> consumers = createConsumers(protocol, consumerCount);
        
        long startTime = System.currentTimeMillis();
        
        // Add new consumer to trigger rebalance
        KafkaConsumer<String, String> newConsumer = createConsumer(protocol);
        newConsumer.subscribe(Arrays.asList("test-topic"));
        
        // Wait for rebalance completion
        awaitRebalanceCompletion(consumers);
        
        long rebalanceTime = System.currentTimeMillis() - startTime;
        
        System.out.println(String.format(
            "Protocol: %s, Consumers: %d, Rebalance Time: %dms", 
            protocol, consumerCount, rebalanceTime
        ));
    }
    
    // Results observed by OSO engineers:
    // Classic Protocol (10 consumers): 8-15 seconds
    // New Protocol (10 consumers): 2-4 seconds
    // Classic Protocol (30 consumers): 25-45 seconds  
    // New Protocol (30 consumers): 3-6 seconds
}

public class FaultToleranceDemo {
    
    public void simulateSlowConsumer() {
        // Classic Protocol Impact
        // - Slow consumer blocks entire group
        // - All partitions stop processing
        // - Timeouts affect healthy consumers
        
        // New Protocol Impact  
        // - Slow consumer only affects its own partitions
        // - Other consumers continue normal processing
        // - Partitions can be migrated away from slow consumer
        
        Properties slowConsumerConfig = new Properties();
        slowConsumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, "consumer");
        
        // Simulate processing delay in rebalance listener
        ConsumerRebalanceListener slowListener = new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                // Simulate slow processing - only affects this consumer
                try {
                    Thread.sleep(10000); // 10 second delay
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
            
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                // Quick assignment
            }
        };
        
        // With new protocol: Other consumers continue processing
        // With classic protocol: All consumers would be blocked
    }
}

// Phase 1: Broker Upgrade (4.x required)
// Ensure all brokers support new protocol

// Phase 2: Consumer Configuration Changes
public class MigrationConfig {
    
    public Properties getClassicConfig() {
        Properties props = new Properties();
        props.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, "classic");
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, 
                  "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "3000");
        return props;
    }
    
    public Properties getNewProtocolConfig() {
        Properties props = new Properties();
        // Single configuration change enables new protocol
        props.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, "consumer");
        
        // Optional: Choose server-side assignor
        props.put("group.remote.assignor", "uniform"); // or "range"
        
        // These configurations no longer supported/needed:
        // - partition.assignment.strategy (moved to broker)
        // - session.timeout.ms (managed by broker)  
        // - heartbeat.interval.ms (managed by broker)
        
        return props;
    }
}

// Phase 3: Rolling Upgrade
public class RollingUpgrade {
    
    public void performRollingUpgrade() {
        // The new protocol supports mixed-mode operation
        // Classic and new protocol consumers can coexist in same group
        // Broker automatically converts protocols for compatibility
        
        // 1. Upgrade consumers one by one
        // 2. Change configuration to group.protocol=consumer
        // 3. Restart consumer
        // 4. Repeat for all consumers in group
        
        // No downtime required - seamless transition
    }
}

This post first appeared on Read More