Consumer Groups and Rebalancing in Kafka with Spring Boot
1. Overview
When running multiple instances of the same service reading from Kafka, you need to know which instance processes which message. Kafka solves this with consumer groups and the rebalancing mechanism.
This article explains how Kafka divides partitions in a group and what situations trigger rebalancing. Then you’ll configure Spring Boot so that rebalancing happens less frequently and doesn’t lose or duplicate data. At the end of the article is a test with Embedded Kafka to verify partition assignment.
2. How Consumer Groups Work
A consumer group is a collection of consumers sharing the same group.id. Kafka ensures:
- Each partition of a topic is assigned to exactly one consumer in the group at any given time.
- A consumer can receive multiple partitions.
- Different groups read independently. Each group has its own offset.
Suppose the orders topic has 4 partitions:
Group "billing" with 2 consumers:
consumer-A -> orders-0, orders-1
consumer-B -> orders-2, orders-3
Group "billing" with 5 consumers:
consumer-A -> orders-0
consumer-B -> orders-1
consumer-C -> orders-2
consumer-D -> orders-3
consumer-E -> (idle, no partition)An important consequence is that the number of partitions limits the maximum parallelism of a group. Adding consumers beyond the number of partitions doesn’t help process faster.
The coordination is handled by one broker, called the group coordinator. Consumers send heartbeats to the coordinator regularly to indicate they’re still alive.
2.1. Dependency
Add Spring for Apache Kafka. Spring Boot 3.x already manages the version:
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<scope>test</scope>
</dependency>2.2. A Simple Listener
@Component
public class OrderListener {
private static final Logger log = LoggerFactory.getLogger(OrderListener.class);
@KafkaListener(topics = "orders", groupId = "billing", concurrency = "2")
public void onOrder(ConsumerRecord<String, String> record) {
log.info("partition={} offset={} order={}",
record.partition(), record.offset(), record.value());
}
}groupId = "billing"puts the consumer in thebillinggroup.concurrency = "2"creates 2 consumers (2 threads) in the same instance. If you run 2 instances of the service, the group will have 4 consumers, each receiving 1 partition.
The log on startup shows the partition assignment result:
billing: partitions assigned: [orders-0, orders-1]
billing: partitions assigned: [orders-2, orders-3]3. What is Rebalancing and When Does It Happen
Rebalancing is the process where a group redistributes partitions among its members. It occurs when:
- A new consumer joins, for example when you scale up or deploy a new instance.
- A consumer leaves gracefully, for example when shutting down with
close(). - A consumer is considered dead: the coordinator doesn’t receive heartbeats within
session.timeout.ms. - A consumer processes too slowly: the gap between two
poll()calls exceedsmax.poll.interval.ms. The consumer leaves the group on its own. - Topic changes: partition count increases, or new topics match a subscribed pattern.
The problem with the old eager protocol is that every consumer must return all its partitions before they can be reassigned. The entire group stops processing during this time, a phenomenon often called “stop-the-world”. With large groups or rolling deployments, rebalancing can repeat continuously.
4. Important Timeout Parameters
spring:
kafka:
bootstrap-servers: localhost:9092
consumer:
group-id: billing
enable-auto-commit: false
max-poll-records: 100
properties:
session.timeout.ms: 45000
heartbeat.interval.ms: 3000
max.poll.interval.ms: 300000| Parameter | Meaning |
|---|---|
session.timeout.ms |
Maximum time without heartbeat before consumer is removed |
heartbeat.interval.ms |
Heartbeat interval, usually ≤ 1/3 of session.timeout.ms |
max.poll.interval.ms |
Maximum gap between two poll() calls |
max-poll-records |
Maximum records per poll() call |
Heartbeats are sent by a separate background thread. So the consumer remains “alive” according to session.timeout.ms even if the main thread is busy. However max.poll.interval.ms only measures the processing thread. If a batch takes too long to process, the consumer will still be kicked out of the group.
A typical error in the logs looks like this:
Member consumer-billing-1 sending LeaveGroup request to coordinator due to
consumer poll timeout has expired. This means the time between subsequent
calls to poll() was longer than the configured max.poll.interval.ms ...
CommitFailedException: Commit cannot be completed since the group has already rebalancedThe formula that must be satisfied:
max-poll-records × processing time per record < max.poll.interval.msNote: Don’t “fix” this by increasing
max.poll.interval.msto several hours. A truly stuck consumer will hold its partitions for a long time without processing anything. Instead, reducemax-poll-records, or move heavy work outside the consume loop.
5. Reducing the Impact of Rebalancing
5.1. Cooperative Rebalancing
With CooperativeStickyAssignor, only partitions that actually need to move are revoked. The remaining consumers continue processing their partitions while rebalancing happens.
spring:
kafka:
consumer:
properties:
partition.assignment.strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignorExample: group has A (orders-0..3), then B joins.
Eager: A returns all 4 partitions -> reassign: A {0,1}, B {2,3}
Cooperative: A only returns {2,3}, keeps processing {0,1} -> B gets {2,3}Note: Don’t switch an active group from eager to cooperative in a single deploy. You need to do a rolling deploy twice. The first time, declare both assignors, e.g.,
CooperativeStickyAssignor,RangeAssignor. Only on the second pass do you removeRangeAssignor.
As of Kafka 4.0, the new rebalance protocol (KIP-848) is GA. You enable it with group.protocol=consumer. Then partition assignment is performed by the broker, rebalancing is incremental by default, and client-side parameters like partition.assignment.strategy and session.timeout.ms are no longer used. If your cluster supports it, this is the first option to consider.
5.2. Static Membership
Each time a pod restarts, the consumer gets a new member.id, triggering rebalancing twice (leaving and rejoining). Static membership assigns a fixed identifier to the consumer:
spring:
kafka:
consumer:
properties:
group.instance.id: billing-${HOSTNAME}
session.timeout.ms: 60000If the consumer returns within session.timeout.ms, the coordinator gives back the exact same partitions without needing to rebalance. This approach works very well with Kubernetes StatefulSets, where HOSTNAME is stable (billing-0, billing-1, etc.).
Note:
group.instance.idmust be unique within the group. Withconcurrency > 1, Spring Kafka automatically appends-nto each child consumer. If two instances use the same ID, the later one will kick out the earlier one (FencedInstanceIdException). Don’t use static membership with Deployments that have random pod names, as you won’t benefit from it.
6. Handling Offsets During Rebalancing
The biggest risk of rebalancing is duplicate processing. Consumer A has processed a record but hasn’t committed it yet, the partition is moved to B, and B reads from the last committed offset again.
Spring Kafka defaults to committing after each batch (AckMode.BATCH) and automatically commits pending offsets before partitions are revoked. If you need to intervene, for example to flush a buffer or log something, use ConsumerAwareRebalanceListener:
@Component
public class OrderRebalanceListener implements ConsumerAwareRebalanceListener {
private static final Logger log = LoggerFactory.getLogger(OrderRebalanceListener.class);
@Override
public void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer,
Collection<TopicPartition> partitions) {
log.info("About to lose partitions {}, flushing in-memory data", partitions);
}
@Override
public void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer,
Collection<TopicPartition> partitions) {
log.info("Committed and returned partitions {}", partitions);
}
@Override
public void onPartitionsAssigned(Consumer<?, ?> consumer,
Collection<TopicPartition> partitions) {
log.info("Received partitions {}", partitions);
}
@Override
public void onPartitionsLost(Consumer<?, ?> consumer,
Collection<TopicPartition> partitions) {
log.warn("Lost partitions {} without committing", partitions);
}
}onPartitionsRevokedBeforeCommitruns before the container commits offsets. This is where you finalize pending work.onPartitionsRevokedAfterCommitruns after offsets have been committed.onPartitionsLostis called when the consumer is removed from the group, for example due to exceedingmax.poll.interval.ms. At this point, the partitions may already belong to another consumer, so don’t commit.
Register the listener with the container factory:
@Configuration
public class KafkaConfig {
@Bean
ConcurrentKafkaListenerContainerFactory<Object, Object> kafkaListenerContainerFactory(
ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
ConsumerFactory<Object, Object> consumerFactory,
OrderRebalanceListener rebalanceListener) {
var factory = new ConcurrentKafkaListenerContainerFactory<Object, Object>();
configurer.configure(factory, consumerFactory);
factory.getContainerProperties().setConsumerRebalanceListener(rebalanceListener);
return factory;
}
}configurer.configure(...) preserves all configuration from application.yml. The next line just adds the rebalance listener.
Note: No matter how well you configure things, Kafka only guarantees at-least-once for regular consumers. Make your processing idempotent, for example check if
orderIdhas already been paid before callingPaymentService. Don’t assume rebalancing will never cause duplicates.
7. Testing Partition Assignment
The following test uses @EmbeddedKafka to verify that two consumers in the same group evenly share 4 partitions:
@EmbeddedKafka(partitions = 4, topics = "orders")
class ConsumerGroupAssignmentTest {
@Test
void twoConsumersInSameGroupShareAllPartitions(EmbeddedKafkaBroker broker) {
Map<String, Object> props = KafkaTestUtils.consumerProps("billing", "false", broker);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
try (var consumerA = new KafkaConsumer<String, String>(props);
var consumerB = new KafkaConsumer<String, String>(props)) {
consumerA.subscribe(List.of("orders"));
consumerB.subscribe(List.of("orders"));
long deadline = System.currentTimeMillis() + 30_000;
while (System.currentTimeMillis() < deadline
&& !(consumerA.assignment().size() == 2 && consumerB.assignment().size() == 2)) {
consumerA.poll(Duration.ofMillis(200));
consumerB.poll(Duration.ofMillis(200));
}
Set<TopicPartition> all = new HashSet<>(consumerA.assignment());
all.addAll(consumerB.assignment());
assertThat(consumerA.assignment()).hasSize(2);
assertThat(consumerB.assignment()).hasSize(2)
.doesNotContainAnyElementsOf(consumerA.assignment());
assertThat(all).hasSize(4);
}
}
}Explanation:
@EmbeddedKafkastarts an in-memory broker, creates theorderstopic with 4 partitions. The broker is injected directly as a test method parameter.- Two
KafkaConsumers share the samegroup.id = "billing". - The loop alternately calls
poll()on both consumers, because consumers only participate in rebalancing insidepoll(). If only consumer A polls, rebalancing won’t complete. Both consumers run on the same thread but are separate objects, so it doesn’t violate the rule that “one consumer should only be used by one thread”. - The assertions verify that each consumer has 2 partitions, they don’t overlap, and together they cover all 4 partitions.
This is also a good way to experiment with different partition.assignment.strategy values. Try adding a third consumer, closing one consumer, then observe how assignment() changes.
8. Common Mistakes
- More consumers than partitions: extra consumers sit idle. Increasing
concurrencydoesn’t make the system faster. - Synchronous processing takes too long in the listener: exceeds
max.poll.interval.ms, causing a rebalancing loop. The consumer is kicked out, comes back, receives the same batch again, then gets kicked out again. - Manual commit in
onPartitionsLost: causesCommitFailedException, or worse, overwrites the new consumer’s offset. - Using the same
group.idfor two different services: the two services will “share” messages instead of each getting all of them. Each service should have its own group. - Rolling deploy with eager protocol on large groups: each pod restart causes two stop-the-world events. Use cooperative protocol, static membership, or Kafka 4.x’s new protocol.
9. Conclusion
Consumer groups allow you to scale Kafka consumption horizontally, with the upper limit being the number of partitions. Rebalancing is the price you pay for that flexibility. It happens when members join, leave, die, or process too slowly.
To minimize rebalancing’s impact, you should:
- Configure
max.poll.interval.msandmax-poll-recordsappropriately. - Use
CooperativeStickyAssignor, orgroup.protocol=consumeron Kafka 4.x. - Enable static membership when your infrastructure allows it.
- Manage the partition lifecycle through
ConsumerAwareRebalanceListener.
Finally, always design your processing to be idempotent, since at-least-once is the default guarantee you need to accept.
Change Data Capture with Debezium and Kafka
Use Debezium to read PostgreSQL's WAL, push every data change to Kafka and consume them with Spring Boot, without modifying the data writing code.
Event Sourcing with Kafka: Save Events Instead of State
Use Kafka as an event log for Order: record immutable events, rebuild state through replay, build read models using projections. Plus the limitations to know before applying.
Kafka Monitoring: consumer lag and key metrics to track
What is consumer lag, how to measure it with CLI, AdminClient and Micrometer, which broker/producer/consumer metrics to track and how to set up reasonable alerts.
Kafka producer: acks, idempotence, and batching
Configure acks, enable.idempotence, batch.size and linger.ms for Kafka producer to prevent message loss, avoid duplicates while achieving high throughput.
Outbox pattern with Kafka: never lose events on database write
Writing to database and sending to Kafka simultaneously easily loses events. This article guides you through implementing the Outbox pattern with Spring Boot 3 and Kafka to keep both systems consistent.