Consumer group y rebalance en Kafka con Spring Boot
1. Descripción general
Cuando ejecutas múltiples instancias del mismo servicio leyendo de Kafka, necesitas saber qué mensaje es procesado por qué instancia. Kafka resuelve esto mediante consumer groups y el mecanismo de rebalance.
Este artículo explica cómo Kafka asigna particiones dentro de un grupo y en qué situaciones ocurre el rebalance. Luego configurarás Spring Boot para que el rebalance sea menos frecuente y no pierda ni duplique datos. Al final hay un test con Embedded Kafka para verificar cómo se asignan las particiones.
2. Cómo funciona el consumer group
Un consumer group es un conjunto de consumidores que comparten el mismo group.id. Kafka garantiza:
- Cada partición del topic se asigna a exactamente un consumidor del grupo en cualquier momento.
- Un consumidor puede recibir múltiples particiones.
- Grupos diferentes leen de forma independiente. Cada grupo tiene su propio offset.
Supongamos que el topic orders tiene 4 particiones:
Grupo "billing" con 2 consumidores:
consumer-A -> orders-0, orders-1
consumer-B -> orders-2, orders-3
Grupo "billing" con 5 consumidores:
consumer-A -> orders-0
consumer-B -> orders-1
consumer-C -> orders-2
consumer-D -> orders-3
consumer-E -> (ocioso, sin particiones)Una consecuencia importante es que el número de particiones limita el paralelismo máximo de un grupo. Añadir consumidores más allá del número de particiones no ayuda a procesar más rápido.
La coordinación la realiza un broker, denominado group coordinator. El consumidor envía heartbeats regularmente al coordinador para indicar que sigue activo.
2.1. Dependencia
Añade Spring for Apache Kafka. Spring Boot 3.x gestiona la versión automáticamente:
<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. Un listener simple
@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("partición={} offset={} orden={}",
record.partition(), record.offset(), record.value());
}
}groupId = "billing"coloca el consumidor en el grupobilling.concurrency = "2"crea 2 consumidores (2 threads) en la misma instancia. Si ejecutas 2 instancias del servicio, el grupo tendrá 4 consumidores, cada uno recibiendo 1 partición.
El log al iniciar muestra el resultado de la asignación:
billing: particiones asignadas: [orders-0, orders-1]
billing: particiones asignadas: [orders-2, orders-3]3. Qué es rebalance y cuándo ocurre
Rebalance es el proceso donde el grupo redistribuye las particiones entre sus miembros. Ocurre cuando:
- Se une un consumidor nuevo, por ejemplo escalas más pods o despliegas una nueva instancia.
- Un consumidor se va normalmente, por ejemplo al apagar y llamar a
close(). - Un consumidor se considera muerto: el coordinador no recibe heartbeat en
session.timeout.ms. - Un consumidor procesa demasiado lentamente: el intervalo entre dos llamadas a
poll()excedemax.poll.interval.ms. El consumidor se retira del grupo automáticamente. - El topic cambia: aumenta el número de particiones, o hay un nuevo topic que coincide con el patrón suscrito.
El problema es que en el protocolo eager antiguo, cada consumidor debe devolver todas sus particiones antes de ser reasignadas. Todo el grupo deja de procesar durante este tiempo, un fenómeno frecuentemente llamado “stop-the-world”. Con grupos grandes o despliegues tipo rolling, el rebalance puede repetirse continuamente.
4. Parámetros de timeout importantes
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| Parámetro | Significado |
|---|---|
session.timeout.ms |
Tiempo máximo sin heartbeat antes de que el consumidor sea expulsado |
heartbeat.interval.ms |
Intervalo de envío de heartbeat, usualmente ≤ 1/3 de session.timeout.ms |
max.poll.interval.ms |
Intervalo máximo entre dos llamadas a poll() |
max-poll-records |
Número máximo de registros por cada poll() |
El heartbeat se envía desde un thread de fondo separado. Por lo tanto, el consumidor sigue “vivo” según session.timeout.ms incluso cuando el thread principal está ocupado. Sin embargo, max.poll.interval.ms solo mide el thread de procesamiento. Si un lote tarda demasiado en procesarse, el consumidor aún será expulsado del grupo.
El error típico en el log se ve así:
El miembro consumer-billing-1 envía solicitud LeaveGroup al coordinador porque
el timeout de poll del consumidor ha expirado. Esto significa que el tiempo entre
llamadas consecutivas a poll() fue mayor que el max.poll.interval.ms configurado ...
CommitFailedException: No se puede completar el commit ya que el grupo ha hecho rebalanceLa fórmula que debe satisfacerse:
max-poll-records × tiempo de procesamiento de 1 registro < max.poll.interval.msNota: No intentes “arreglarlo” aumentando
max.poll.interval.msa varias horas. Un consumidor realmente congelado mantendrá las particiones durante mucho tiempo sin que nadie las procese. Reducemax-poll-records, o mueve las tareas pesadas fuera del hilo de consumo.
5. Reducir el impacto del rebalance
5.1. Rebalance cooperativo
Con CooperativeStickyAssignor, solo las particiones que realmente necesitan cambiar se recuperan. Los otros consumidores continúan procesando sus particiones mientras ocurre el rebalance.
spring:
kafka:
consumer:
properties:
partition.assignment.strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignorEjemplo: el grupo tiene A (orders-0..3), luego se une B.
Eager: A devuelve las 4 particiones -> reasigna: A {0,1}, B {2,3}
Cooperativo: A solo devuelve {2,3}, sigue procesando {0,1} continuamente -> B recibe {2,3}Nota: No cambies directamente un grupo en ejecución de eager a cooperative en un único despliegue. Necesitas rolling deploy dos veces. En el primer despliegue, declara ambos asignadores, por ejemplo
CooperativeStickyAssignor,RangeAssignor. En el segundo despliegue, eliminaRangeAssignor.
Desde Kafka 4.0, el nuevo protocolo de rebalance (KIP-848) es GA. Lo activas con group.protocol=consumer. Entonces el broker es quien realiza la asignación, el rebalance es incremental por defecto, y parámetros como partition.assignment.strategy y session.timeout.ms del lado cliente ya no se usan. Si tu cluster lo soporta, esta es la primera opción a considerar.
5.2. Static membership
Cada vez que reinicia un pod, el consumidor recibe un nuevo member.id, lo que dispara rebalance dos veces (al salir y volver a entrar). Static membership asigna una identidad fija al consumidor:
spring:
kafka:
consumer:
properties:
group.instance.id: billing-${HOSTNAME}
session.timeout.ms: 60000Si el consumidor vuelve dentro del tiempo session.timeout.ms, el coordinador le asigna nuevamente exactamente sus particiones anteriores sin necesidad de rebalance. Esto es muy adecuado para Kubernetes StatefulSet, donde HOSTNAME es estable (billing-0, billing-1…).
Nota: El
group.instance.iddebe ser único dentro del grupo. Conconcurrency > 1, Spring Kafka añade automáticamente un sufijo-na cada consumidor. Si dos instancias usan el mismo ID, la instancia que llega después “expulsará” a la anterior (FencedInstanceIdException). No uses static membership con Deployment que tenga nombres de pods aleatorios, porque no obtendrás beneficio.
6. Manejo de offset durante el rebalance
El mayor riesgo del rebalance es el procesamiento duplicado. El consumidor A ha procesado un registro pero aún no ha hecho commit, la partición se transfiere a B, y B lee nuevamente desde el último offset confirmado.
Spring Kafka por defecto hace commit después de cada lote (AckMode.BATCH) y automáticamente confirma los offsets pendientes antes de que se recupere una partición. Si necesitas intervenir, por ejemplo para vaciar un buffer o escribir en log, usas 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("A punto de perder particiones {}, vaciar datos en memoria", partitions);
}
@Override
public void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer,
Collection<TopicPartition> partitions) {
log.info("Se ha confirmado y devuelto la partición {}", partitions);
}
@Override
public void onPartitionsAssigned(Consumer<?, ?> consumer,
Collection<TopicPartition> partitions) {
log.info("Asignadas particiones {}", partitions);
}
@Override
public void onPartitionsLost(Consumer<?, ?> consumer,
Collection<TopicPartition> partitions) {
log.warn("Particiones {} perdidas sin confirmar", partitions);
}
}onPartitionsRevokedBeforeCommitse ejecuta antes de que el container confirme los offsets. Aquí es donde finalizas el trabajo pendiente.onPartitionsRevokedAfterCommitse ejecuta después de que los offsets han sido confirmados.onPartitionsLostse llama cuando el consumidor es expulsado del grupo, por ejemplo por excedermax.poll.interval.ms. En este caso, la partición puede ya ser de otro consumidor, así que no hagas commit.
Registra el listener en la factory del container:
@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(...) mantiene toda la configuración de application.yml. La siguiente línea solo añade el listener de rebalance.
Nota: No importa cuán bien configures, Kafka solo garantiza at-least-once para consumidores normales. Haz que tu procesamiento sea idempotente, por ejemplo verificando que el
orderIdya fue pagado antes de llamar aPaymentService. No asutas que el rebalance nunca causará duplicados.
7. Prueba de la asignación de particiones
Esta prueba usa @EmbeddedKafka para verificar que dos consumidores del mismo grupo se dividen las 4 particiones equitativamente:
@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);
}
}
}Explicación:
@EmbeddedKafkainicia un broker en memoria, crea el topicorderscon 4 particiones. El broker se inyecta directamente como parámetro del método de prueba.- Dos
KafkaConsumercomparten el mismogroup.id = "billing". - El bucle alterna llamadas a
poll()en ambos consumidores, porque el consumidor solo participa en rebalance dentro depoll(). Si solo A hace poll, el rebalance no se completará. Ambos consumidores se ejecutan en el mismo thread pero son objetos separados, así que no violas la regla de “un consumidor solo es usado por un thread”. - Las aserciones verifican que cada consumidor tenga 2 particiones, que no se superpongan, y que en total cubran las 4 particiones.
Esta también es una buena forma de experimentar con diferentes partition.assignment.strategy. Intenta añadir un tercer consumidor, cierra uno, y observa cómo cambia assignment().
8. Errores comunes
- Más consumidores que particiones: los consumidores excedentes estarán ociosos. Aumentar
concurrencyno acelera el sistema. - Procesamiento síncrono demasiado lento en el listener: excede
max.poll.interval.mscausando un bucle de rebalance. El consumidor es expulsado, vuelve, recibe el mismo lote, y es expulsado de nuevo. - Hacer commit manual en
onPartitionsLost: causaCommitFailedException, o peor, sobrescribe el offset del nuevo consumidor. - Usar el mismo
group.idpara dos servicios diferentes: los dos servicios “se compartirán” los mensajes en lugar de que cada uno reciba todos. Cada servicio independiente debe tener su propio grupo. - Rolling deploy con protocolo eager en grupos grandes: cada reinicio de pod causa dos paradas. Usa cooperative, static membership, o el nuevo protocolo de Kafka 4.x.
9. Conclusión
El consumer group te permite escalar la lectura de Kafka horizontalmente, con un límite superior dado por el número de particiones. El rebalance es el precio por esa flexibilidad. Ocurre cuando miembros entran, salen, mueren o procesan demasiado lentamente.
Para que el rebalance cause menos daño, deberías hacer lo siguiente:
- Configurar razonablemente
max.poll.interval.msymax-poll-records. - Usar
CooperativeStickyAssignor, ogroup.protocol=consumeren Kafka 4.x. - Habilitar static membership cuando la infraestructura lo permita.
- Manejar el ciclo de vida de las particiones mediante
ConsumerAwareRebalanceListener.
Finalmente, siempre diseña tu procesamiento para ser idempotente, porque at-least-once es la garantía por defecto que debes aceptar.
¿Te gusta este tipo de explicación? Cris The Dev publica un vídeo corto cada día en YouTube y TikTok.
Más sobre este tema
Change Data Capture con Debezium y Kafka
Usa Debezium para capturar cambios de PostgreSQL, envíalos a Kafka y consume con Spring Boot, eliminando el riesgo del dual-write.
Event sourcing con Kafka: guardar eventos en lugar de estado
Usa Kafka como event log para Order: registra eventos inmutables, reconstruye el estado con replay, crea read models con projection. Incluye limitaciones que debes conocer antes de aplicar.
Kafka producer: acks, idempotencia y batching
Configura acks, enable.idempotence, batch.size y linger.ms en Kafka producer para no perder mensajes, evitar duplicados y alcanzar alto throughput.
Monitoreo de Kafka: consumer lag y métricas clave
Qué es el consumer lag, cómo medirlo con CLI, AdminClient y Micrometer, qué métricas del broker/productor/consumidor debes vigilar, y cómo establecer alertas efectivas.
Outbox pattern con Kafka: no pierdas eventos al escribir
Escribir en base de datos y enviar a Kafka simultáneamente pierde eventos fácilmente. Este artículo enseña a implementar Outbox pattern con Spring Boot 3 y Kafka para mantener coherencia.