Kafka - Consumer Group Rebalancing & Strategies

A rebalance is the mechanism Kafka uses to redistribute partitions across the consumers in a consumer group. Every consumer group has a group coordinator: a broker that manages group membership and triggers rebalances when the group state changes.

Rebalances are triggered when:

• A new consumer joins the group (e.g., a new instance starts up).
• An existing consumer leaves the group (graceful shutdown via consumer.close()).
• A consumer is considered dead: it fails to send heartbeats within session.timeout.ms or fails to call poll() within max.poll.interval.ms.
• New partitions are added to a subscribed topic.
• The subscription pattern matches a newly created topic (when using regex subscription).

During a rebalance the group coordinator collects partition assignments from all active consumers, then uses a configurable partition assignor to compute the new mapping. The result is sent back to each consumer so it knows which partitions to read from.

The eager rebalance protocol is the original Kafka approach. When a rebalance begins, every consumer in the group revokes all of its partitions before any partitions are reassigned. This causes a complete pause of consumption across the entire group.

The two built-in eager assignors are:

• RangeAssignor (default before Kafka 3.0): Assigns partitions on a per-topic basis. Partitions of each topic are divided among consumers in lexicographic order by consumer id. This can lead to uneven distribution when the partition count is not evenly divisible.
• RoundRobinAssignor: Assigns all partitions across all subscribed topics in a round-robin fashion. Produces a more balanced assignment than RangeAssignor when consumers subscribe to the same set of topics.

// Configure RoundRobinAssignor (eager protocol)
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processing-group");
props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
          "org.apache.kafka.clients.consumer.RoundRobinAssignor");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
          "org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
          "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("orders", "payments"));

Cooperative rebalancing (introduced in Kafka 2.4) eliminates the stop-the-world pause by only revoking the partitions that actually need to move to a different consumer. Consumers that keep their existing partitions continue processing without interruption.

The cooperative protocol uses a two-phase approach:

• Phase 1: The assignor computes the ideal assignment and identifies which partitions need to migrate. Only those partitions are revoked from their current owners. All other partitions continue being consumed normally.
• Phase 2: A second rebalance is triggered automatically. The revoked partitions are now unowned and get assigned to their new consumers.

The built-in cooperative assignor is CooperativeStickyAssignor, which combines the incremental protocol with sticky assignment: it tries to preserve existing partition-to-consumer mappings as much as possible.

// Configure CooperativeStickyAssignor (incremental protocol)
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processing-group");
props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
          "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
          "org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
          "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("orders", "payments"));

When migrating from an eager assignor to a cooperative assignor in a running consumer group, you must perform a rolling upgrade:

• First, configure all consumers with both the old and new assignors (e.g., RangeAssignor, CooperativeStickyAssignor).
• Perform a rolling restart so every consumer advertises support for both protocols.
• Then remove the old eager assignor from the configuration and do another rolling restart.

By default, every time a consumer restarts it gets a new member.id from the group coordinator. This means even a brief restart triggers a full rebalance. Static group membership (introduced in Kafka 2.3) solves this by assigning a persistent identity to each consumer via group.instance.id.

When a consumer with a group.instance.id disconnects, the group coordinator does not trigger a rebalance immediately. Instead, it waits for session.timeout.ms to expire. If the consumer reconnects with the same instance id before the timeout, it silently rejoins and reclaims its previous partition assignment without any rebalance.

// Static group membership — avoid rebalance on restart
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processing-group");
props.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, "consumer-host-1"); // unique per instance
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 60000);           // 60s window for restart
props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
          "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
          "org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
          "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("orders"));

The ConsumerRebalanceListener interface lets you execute custom logic when partitions are revoked from or assigned to a consumer during a rebalance. The two key callbacks are:

• onPartitionsRevoked(Collection<TopicPartition>): Called before partitions are taken away. This is where you should commit offsets, flush buffers, or close resources tied to those partitions.
• onPartitionsAssigned(Collection<TopicPartition>): Called after new partitions have been assigned. Use this to initialize state, seek to a specific offset, or prepare resources for the new partitions.

consumer.subscribe(List.of("orders"), new ConsumerRebalanceListener() {

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        System.out.println("Partitions revoked: " + partitions);

        // Commit current offsets before losing ownership
        consumer.commitSync();

        // Flush any buffered state (e.g., batch inserts)
        flushPendingWrites();
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        System.out.println("Partitions assigned: " + partitions);

        // Optionally seek to a stored offset (e.g., from an external store)
        for (TopicPartition tp : partitions) {
            Long savedOffset = externalStore.getOffset(tp);
            if (savedOffset != null) {
                consumer.seek(tp, savedOffset);
            }
        }
    }
});

With the cooperative protocol there is an additional callback:

• onPartitionsLost(Collection<TopicPartition>): Called when partitions are lost unexpectedly (e.g., consumer was fenced). Unlike onPartitionsRevoked, the partitions may already be owned by another consumer, so you should not commit offsets here.

consumer.subscribe(List.of("orders"), new ConsumerRebalanceListener() {

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // Safe to commit — we still own these partitions
        consumer.commitSync(currentOffsets(partitions));
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // Initialize resources for newly assigned partitions
        initializeState(partitions);
    }

    @Override
    public void onPartitionsLost(Collection<TopicPartition> partitions) {
        // Partitions already moved — do NOT commit offsets
        // Clean up local state only
        cleanupState(partitions);
    }
});

Kafka provides four built-in partition assignors. Each implements a different strategy for distributing partitions across the consumers in a group.

# Example: setting the assignor in consumer properties
partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor

A rebalance storm occurs when repeated rebalances fire in quick succession, preventing the consumer group from making progress. This is one of the most common operational issues in Kafka consumer deployments.

Common causes and their fixes:

• Slow message processing: If processing a batch takes longer than max.poll.interval.ms (default 300 seconds), the consumer is evicted and triggers a rebalance. Fix: increase max.poll.interval.ms or reduce max.poll.records to get smaller batches.
• Long GC pauses: Garbage collection pauses exceeding session.timeout.ms cause the coordinator to mark the consumer as dead. Fix: tune JVM GC settings, increase session.timeout.ms.
• Frequent deploys without static membership: Rolling deployments cause a rebalance on every pod restart. Fix: configure group.instance.id for static membership.
• Network issues: Transient network partitions prevent heartbeats from reaching the coordinator. Fix: verify network stability, increase session.timeout.ms.
• Too many consumers: More consumers than partitions means idle consumers. Scaling up and down repeatedly triggers rebalances. Fix: ensure consumer count does not exceed partition count.

Key configuration properties to tune:

# Tuning to prevent rebalance storms
max.poll.interval.ms=600000
max.poll.records=100
session.timeout.ms=60000
heartbeat.interval.ms=10000

Spring Kafka provides convenient configuration for partition assignment strategies through application properties and programmatic configuration.

# application.properties — Spring Boot Kafka consumer configuration
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=order-processing-group
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

# Set cooperative sticky assignor
spring.kafka.consumer.properties.partition.assignment.strategy=\
  org.apache.kafka.clients.consumer.CooperativeStickyAssignor

# Static membership for Kubernetes deployments
spring.kafka.consumer.properties.group.instance.id=${HOSTNAME}
spring.kafka.consumer.properties.session.timeout.ms=60000

For programmatic configuration using ConcurrentKafkaListenerContainerFactory:

@Configuration
public class KafkaConsumerConfig {

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processing-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
                  List.of(CooperativeStickyAssignor.class));
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 60000);
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 600000);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
        return new DefaultKafkaConsumerFactory<>(props);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3); // number of consumer threads
        return factory;
    }
}

Using @KafkaListener with a rebalance listener:

@Component
public class OrderConsumer {

    @KafkaListener(topics = "orders", groupId = "order-processing-group")
    public void consume(ConsumerRecord<String, String> record,
                        Acknowledgment acknowledgment) {
        processOrder(record.value());
        acknowledgment.acknowledge();
    }
}

@Component
public class RebalanceLogger implements ConsumerAwareRebalanceListener {

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        log.info("Partitions revoked: {}", partitions);
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        log.info("Partitions assigned: {}", partitions);
    }

    @Override
    public void onPartitionsLost(Collection<TopicPartition> partitions) {
        log.warn("Partitions lost: {}", partitions);
    }
}