Consumer group y rebalance en Kafka: por qué se detienen

1. Descripción general

Tienes un servicio que lee mensajes de Kafka y funciona de manera estable durante muchos días. Un día, el retraso del consumidor se dispara. El registro está lleno de líneas como Revoking previously assigned partitions y CommitFailedException. El culpable suele ser el rebalance del consumer group.

Este artículo explica cómo un consumer group divide las particiones, cuándo ocurre un rebalance y por qué hace que los consumidores se “congelen”. Después de leerlo, sabrás cómo configurar Spring Kafka para reducir la cantidad de rebalances y el tiempo de detención en cada uno.

2. Cómo funciona un consumer group

Un consumer group es un conjunto de consumidores que comparten un group.id. Kafka asigna las particiones del tema a los miembros del grupo. Cada partición se asigna a un único consumidor en un momento dado.

Supongamos que el tema orders tiene 6 particiones. Ejecutas 3 instancias de order-service:

order-service-1  -> orders-0, orders-1
order-service-2  -> orders-2, orders-3
order-service-3  -> orders-4, orders-5

Si ejecutas 7 instancias, una instancia quedará sin hacer nada, porque no hay más particiones para asignar.

Primero, añade la dependencia:

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

Un escucha simple en Spring Boot 3.x:

@Component
public class OrderListener {

    private static final Logger log = LoggerFactory.getLogger(OrderListener.class);

    @KafkaListener(topics = "orders", groupId = "order-service")
    public void onOrder(ConsumerRecord<String, String> record) {
        log.info("Se recibió el pedido {} de la partición {}", record.key(), record.partition());
    }
}

La anotación @KafkaListener crea un contenedor de escucha. Dentro del contenedor hay un hilo que ejecuta un bucle poll(). Todas las instancias que usan groupId = "order-service" participarán en el mismo grupo. Un broker actúa como coordinador del grupo que gestiona la lista de miembros y la asignación de particiones.

3. Qué es rebalance y cuándo ocurre

Rebalance es el proceso en el que el coordinador del grupo reasigna las particiones a los miembros. Las situaciones que causan un rebalance:

1. Un nuevo consumidor se une: escalado, despliegue de una nueva instancia.
2. Un consumidor se retira del grupo voluntariamente: apagado, reinicio gradual.
3. Un consumidor se considera muerto porque no envía latidos dentro de session.timeout.ms. El valor predeterminado es 45 segundos.
4. Un consumidor no llama a poll() dentro de max.poll.interval.ms. El valor predeterminado es 5 minutos.
5. El número de particiones del tema cambia, o la suscripción según patrón coincide con nuevos temas.

Los dos mecanismos de tiempo de espera en los puntos 3 y 4 son diferentes y suelen confundirse:

Configuración	Predeterminado	Qué verifica
heartbeat.interval.ms	3000	Intervalo de envío de latidos (desde un hilo de fondo)
session.timeout.ms	45000	Si el proceso sigue vivo
max.poll.interval.ms	300000	Si el hilo de procesamiento está atrapado

Los latidos se envían desde un hilo de fondo separado. Por lo tanto, el proceso sigue “vivo” cuando el hilo de procesamiento está atrapado. Por el contrario, max.poll.interval.ms detecta correctamente el caso en el que el código de procesamiento de mensajes es demasiado lento.

4. Por qué los consumidores se detienen

4.1. Rebalance entusiasta: detiene todo el grupo

Con el protocolo tradicional y asignadores de tipo entusiasta (RangeAssignor, RoundRobinAssignor, StickyAssignor), cada rebalance ocurre así:

1. Todos los consumidores revocan todas las particiones que mantienen.
2. Todos los consumidores reenvían una solicitud JoinGroup y se esperan mutuamente.
3. El líder del grupo calcula un nuevo plan de asignación, luego el coordinador devuelve el resultado.
4. Los consumidores reciben las nuevas particiones y continúan procesando.

Mientras ocurren los pasos 1–3, nadie en el grupo procesa mensajes. Solo una nueva instancia que se inicia puede detener todo el grupo. Este es el fenómeno “stop-the-world”.

4.2. Procesamiento demasiado lento: exceder max.poll.interval.ms

La causa más común en la práctica es que el código sea lento. El siguiente ejemplo llama a un servicio externo que tarda aproximadamente 1 segundo por mensaje:

@KafkaListener(topics = "orders", groupId = "order-service")
public void onOrder(ConsumerRecord<String, String> record) throws InterruptedException {
    paymentService.charge(record.value()); // supongamos que tarda ~1 segundo
}

Por defecto, max.poll.records = 500. Una única llamada a poll() puede devolver 500 registros, es decir, aproximadamente 500 segundos de procesamiento. Este número excede los 300 segundos de max.poll.interval.ms. Cuando esto sucede, el consumidor se retira del grupo automáticamente, y verás un registro como este:

[Consumer clientId=consumer-order-service-1, groupId=order-service] consumer poll timeout has expired.
This means the time between subsequent calls to poll() was longer than the configured
max.poll.interval.ms, which typically implies that the poll loop is spending too much time
processing messages. You can address this either by increasing max.poll.interval.ms or by
reducing the maximum size of batches returned in poll() with max.poll.records.

Después, el comando de confirmación de offset falla:

org.apache.kafka.clients.consumer.CommitFailedException: Offset commit cannot be completed
since the consumer is not part of an active group for auto partition assignment; it is likely
that the consumer was kicked out of the group.

Esto desencadena una serie de problemas:

• La partición de este consumidor se asigna a otro.
• El offset no se confirmó, por lo que el nuevo receptor de la partición reprocesa esos mensajes.
• El consumidor antiguo se reúne nuevamente, causando otro rebalance.
• Si el siguiente lote sigue siendo lento, el ciclo se repite indefinidamente. El retraso crece sin cesar.

5. Cómo solucionar

5.1. Ajustar el tamaño del lote y el tiempo de encuesta

Regla importante: max.poll.records × tiempo de procesamiento por registro debe ser significativamente menor que max.poll.interval.ms.

spring:
  kafka:
    consumer:
      group-id: order-service
      max-poll-records: 50
      properties:
        max.poll.interval.ms: 120000

Con 50 registros y 1 segundo cada uno, un lote tarda aproximadamente 50 segundos. Este número está seguro dentro del límite de 120 segundos.

Nota: No aumentes max.poll.interval.ms a varias horas para “apagar el fuego”. Este valor también es el tiempo máximo para que Kafka detecte que un consumidor está realmente congelado. Además, un rebalance debe esperar a que el consumidor más lento vuelva a llamar a poll(), por lo que un valor grande hace que el rebalance se alargue en consecuencia.

Si cada mensaje tarda mucho en procesarse de todos modos, por ejemplo, cuando se llama a una API lenta, debes establecer un tiempo de espera para la llamada externa. Otra opción es enviar el mensaje a una cola interna, en lugar de mantener el hilo de encuesta ocupado durante mucho tiempo.

5.2. Rebalance cooperativo

CooperativeStickyAssignor utiliza rebalance incremental. Los consumidores solo revocan las particiones que realmente deben transferirse, mientras que las demás particiones siguen siendo procesadas normalmente.

spring:
  kafka:
    consumer:
      properties:
        partition.assignment.strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignor

Cuando añades order-service-4 al grupo de 3 instancias de la sección 2, solo unas pocas particiones se transfieren a la nueva instancia:

order-service-1  -> orders-0, orders-1        (sin cambios, no se detiene)
order-service-2  -> orders-2                  (pierde orders-3)
order-service-3  -> orders-4                  (pierde orders-5)
order-service-4  -> orders-3, orders-5

Nota: No puedes cambiar directamente de entusiasta a cooperativo en un grupo que se está ejecutando. Necesitas un despliegue gradual en dos fases. En la primera fase, configura ambos: CooperativeStickyAssignor,RangeAssignor. En la segunda fase, elimina RangeAssignor. Si los miembros no tienen un asignador en común, el grupo reportará un error.

5.3. Membresía estática

Cada vez que haces un reinicio gradual, una instancia se retira del grupo y luego regresa. El resultado es dos rebalances por pod. La membresía estática resuelve esto. Asignas a cada instancia un identificador fijo a través de group.instance.id:

spring:
  kafka:
    consumer:
      properties:
        group.instance.id: ${HOSTNAME}
        session.timeout.ms: 60000

Un consumidor con group.instance.id no enviará LeaveGroup al apagarse. Si regresa dentro de session.timeout.ms con el mismo id, el coordinador devuelve exactamente las mismas particiones que tenía sin rebalance.

Nota: HOSTNAME debe ser estable entre reinicios, por ejemplo, el nombre del pod de un StatefulSet en Kubernetes. Con un Deployment normal, el nombre del pod cambia cada vez, por lo que la membresía estática no tiene efecto. Dos instancias ejecutándose simultáneamente con el mismo id se expulsarán mutuamente del grupo (FencedInstanceIdException). Cuando concurrency > 1, Spring Kafka añade automáticamente sufijos -0, -1… al id para cada consumidor hijo.

La compensación de este enfoque: cuando una instancia muere realmente, sus particiones no serán procesadas por nadie hasta que expire session.timeout.ms.

5.4. Nuevo protocolo de consumer group

A partir de Kafka 4.0, el nuevo protocolo de rebalance (KIP-848) está listo para producción. Lo activas con group.protocol=consumer. La asignación de particiones se transfiere completamente al broker y siempre ocurre de forma incremental. Los consumidores ya no tienen que esperarse mutuamente a través de una “barrera” JoinGroup compartida.

Con este protocolo, partition.assignment.strategy en el lado del cliente ya no se usa. En su lugar, el broker elige el asignador (la configuración del lado del cliente es group.remote.assignor). Si tu cluster y cliente son lo suficientemente nuevos, esto es una opción a considerar. Debes verificar la versión de kafka-clients que Spring Boot está extrayendo antes de habilitarlo.

6. Manejo cuando se revocan las particiones

Sin importar cuánto optimices, el rebalance seguirá ocurriendo. Lo importante es confirmar los offsets ya procesados antes de perder las particiones, para reducir el reprocesamiento.

Spring Kafka confirma automáticamente los offsets pendientes antes de que se revoque una partición. Si usas KafkaConsumer directamente, debes hacerlo manualmente usando ConsumerRebalanceListener:

public class CommitOnRevokeListener implements ConsumerRebalanceListener {

    private final Consumer<String, String> consumer;
    private final Map<TopicPartition, OffsetAndMetadata> pending = new ConcurrentHashMap<>();

    public CommitOnRevokeListener(Consumer<String, String> consumer) {
        this.consumer = consumer;
    }

    public void markProcessed(ConsumerRecord<String, String> record) {
        pending.put(new TopicPartition(record.topic(), record.partition()),
                new OffsetAndMetadata(record.offset() + 1));
    }

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        Map<TopicPartition, OffsetAndMetadata> toCommit = new HashMap<>();
        for (TopicPartition tp : partitions) {
            OffsetAndMetadata offset = pending.remove(tp);
            if (offset != null) {
                toCommit.put(tp, offset);
            }
        }
        if (!toCommit.isEmpty()) {
            consumer.commitSync(toCommit);
        }
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // Puedes recargar el estado local para las nuevas particiones aquí
    }
}

Explicación de cada parte:

• markProcessed() registra el offset siguiente a leer, es decir, offset + 1, después de que se procese un registro.
• onPartitionsRevoked() se ejecuta en el mismo hilo de encuesta, justo antes de que se tomen las particiones. Solo confirma los offsets de las particiones que se revocan. Este es un punto clave con rebalance cooperativo, porque solo se revocan algunas particiones.
• commitSync() se bloquea hasta que el broker confirme, asegurando que el offset se guarde antes de que otro consumidor reciba la partición.

Registras el escucha al suscribirte: consumer.subscribe(List.of("orders"), listener).

Verifica este comportamiento con MockConsumer disponible en kafka-clients:

class CommitOnRevokeListenerTest {

    @Test
    void shouldCommitOnlyRevokedPartitions() {
        MockConsumer<String, String> consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
        TopicPartition p0 = new TopicPartition("orders", 0);
        TopicPartition p1 = new TopicPartition("orders", 1);
        consumer.assign(List.of(p0, p1));

        CommitOnRevokeListener listener = new CommitOnRevokeListener(consumer);
        listener.markProcessed(new ConsumerRecord<>("orders", 0, 41L, "order-1", "{}"));
        listener.markProcessed(new ConsumerRecord<>("orders", 1, 7L, "order-2", "{}"));

        listener.onPartitionsRevoked(List.of(p0));

        Map<TopicPartition, OffsetAndMetadata> committed = consumer.committed(Set.of(p0, p1));
        assertThat(committed.get(p0).offset()).isEqualTo(42L);
        assertThat(committed.get(p1)).isNull();
    }
}

La prueba simula que solo orders-0 se revoca. El resultado esperado es que el offset 42 se confirme para orders-0. Mientras que orders-1 sigue siendo retenido, por lo que aún no hay ningún offset confirmado para él.

Nota: El código dentro de onPartitionsRevoked() debe ser rápido. Este callback se ejecuta durante el rebalance. Si es lento, todo el grupo debe esperar.

7. Errores comunes

• Procesamiento asincrónico pero se sigue confirmando según la encuesta. Envías un registro a un ExecutorService, luego el contenedor confirma el offset inmediatamente después de poll(). Si ocurre un rebalance, el mensaje en proceso puede perderse. Solo confirma offsets de registros completamente procesados.
• Confundir session.timeout.ms con max.poll.interval.ms. Aumentar el tiempo de sesión no ayuda si la causa es el procesamiento lento.
• Establecer concurrency mayor que el número de particiones. Los consumidores adicionales no tienen trabajo. Solo hacen que el grupo sea más grande y el rebalance más lento.
• GC pause largo o CPU limitado en el contenedor. El hilo de latido no puede ejecutarse a tiempo, el consumidor se expulsa del grupo aunque el código no sea lento. Verifica las métricas de JVM y los límites de CPU.
• No diseñar el procesamiento de forma idempotente. El rebalance siempre puede causar que algunos mensajes se procesen más de una vez. No asumir “exactamente una vez” a menos que uses transacciones o tengas un mecanismo de deduplicación, como verificar si una orderId ya se ha procesado.

8. Conclusión

El rebalance es un mecanismo necesario para que un consumer group se escale automáticamente y tolere fallos. El precio es que los consumidores deben pausarse temporalmente. La causa más común por la que los consumidores se “detienen” continuamente es que el bucle poll() excede max.poll.interval.ms. En ese caso, el consumidor se expulsa del grupo, la confirmación falla y el rebalance se repite indefinidamente.

Para minimizar esto, puedes:

• Mantener cada lote pequeño con max.poll.records.
• Cambiar a CooperativeStickyAssignor o al nuevo protocolo de consumer group.
• Usar membresía estática para despliegues con identificadores estables.
• Confirmar offsets en el momento correcto en onPartitionsRevoked().

Finalmente, siempre diseña el procesamiento de mensajes de forma idempotente. El rebalance no puede eliminarse por completo, solo puede ocurrir con menos frecuencia y causar menos daño.