Consumer group e rebalance no Kafka: por que o consumer para

1. Visão geral

Você tem um serviço que lê mensagens do Kafka e roda estável há vários dias. Um dia o lag do consumer dispara. O log fica repleto de linhas Revoking previously assigned partitions e CommitFailedException. O culpado geralmente é o rebalance do consumer group.

Este artigo explica como um consumer group divide as partições, quando ocorre o rebalance e por que faz o consumer “travar”. Ao final, você saberá como configurar o Spring Kafka para reduzir o número de rebalances e minimizar o tempo de cada pausa.

2. Como funciona um consumer group

Um consumer group é um conjunto de consumers que compartilham o mesmo group.id. Kafka distribui as partições do topic entre os membros do group. Cada partição é atribuída a apenas um consumer por vez.

Suponha que o topic orders tenha 6 partições. Você executa 3 instâncias do order-service:

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

Se você executar 7 instâncias, uma delas ficará ociosa, pois não há mais partições para atribuir.

Primeiro, adicione a dependência:

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

Um listener simples no 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("Recebeu ordem {} da partição {}", record.key(), record.partition());
    }
}

A annotation @KafkaListener cria um listener container. Dentro do container há uma thread que executa um loop poll(). Toda instância usando groupId = "order-service" fará parte do mesmo group. Um broker atua como group coordinator gerenciando a lista de membros e a distribuição de partições.

3. O que é rebalance e quando acontece

Rebalance é o processo em que o group coordinator redistribui as partições entre os membros. Situações que causam rebalance:

1. Um novo consumer entra: scale up, implantação de nova instância.
2. Consumer sai do group voluntariamente: shutdown, rolling restart.
3. Consumer é considerado morto por não enviar heartbeat dentro de session.timeout.ms. O padrão é 45 segundos.
4. Consumer não chama poll() dentro de max.poll.interval.ms. O padrão é 5 minutos.
5. O número de partições do topic muda, ou subscription por pattern começa a corresponder a novos topics.

Os dois mecanismos de timeout nos itens 3 e 4 são diferentes e frequentemente confundidos:

Configuração	Padrão	O que verifica
heartbeat.interval.ms	3000	Intervalo de envio de heartbeat (de uma thread de background)
session.timeout.ms	45000	Se o processo ainda está vivo
max.poll.interval.ms	300000	Se a thread de processamento está travada

O heartbeat é enviado por uma thread de background separada. Portanto, o processo ainda está “vivo” quando a thread de processamento está travada. Por outro lado, max.poll.interval.ms detecta corretamente quando o código de processamento da sua mensagem é muito lento.

4. Por que o consumer para

4.1. Eager rebalance: para todo o group

Com o protocolo tradicional e com assignors do tipo eager (RangeAssignor, RoundRobinAssignor, StickyAssignor), cada rebalance ocorre da seguinte forma:

1. Todos os consumers revogam todas as partições que possuem.
2. Todos os consumers reenvia as solicitações JoinGroup e se esperam.
3. O líder do group calcula um novo plano de distribuição, e o coordinator envia o resultado de volta.
4. Os consumers recebem novas partições e continuam o processamento.

Durante os passos 1-3, ninguém no group processa mensagens. Apenas uma nova instância sendo iniciada já para todo o group. Este é o fenômeno de “stop-the-world”.

4.2. Processamento muito lento: excedendo max.poll.interval.ms

A causa mais comum na prática é o código de processamento lento. O exemplo a seguir chama um serviço externo que leva 1 segundo por mensagem:

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

O padrão é max.poll.records = 500. Uma chamada a poll() pode retornar 500 registros, ou seja, cerca de 500 segundos de processamento. Este número excede os 300 segundos de max.poll.interval.ms. Nesse caso, o consumer sai do group por si só, e você verá um log 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.

Depois, o comando de commit de offset falha:

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.

As consequências trazem uma série de problemas:

• As partições deste consumer são atribuídas a outro.
• O offset não confirmado, então quem recebe a partição reprocessa essas mensagens.
• O consumer antigo se junta novamente, causando mais um rebalance.
• Se o próximo batch continuar lento, o loop se repete indefinidamente. O lag aumenta continuamente.

5. Como corrigir

5.1. Ajuste o tamanho do batch e o tempo de poll

A regra a lembrar: max.poll.records × tempo de processamento por registro deve 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

Com 50 registros e 1 segundo cada um, um batch leva cerca de 50 segundos. Este número está seguramente dentro do limite de 120 segundos.

Nota: Não aumente max.poll.interval.ms para algumas horas para “apagar incêndios”. Este valor também é o tempo máximo para o Kafka detectar um consumer realmente travado. Além disso, um rebalance deve esperar o consumer mais lento chamar poll() novamente, então um valor grande faz o rebalance durar mais.

Se cada mensagem é inerentemente lenta de processar, por exemplo, chamar uma API lenta, você deve definir um timeout para a chamada externa. Outra forma é enviar a mensagem para uma fila interna, em vez de manter a thread poll ocupada por muito tempo.

5.2. Rebalancing cooperativo

CooperativeStickyAssignor usa rebalance incremental. O consumer revoga apenas as partições que realmente precisam ser movidas, enquanto outras partições continuam sendo processadas normalmente.

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

Quando você adiciona order-service-4 ao grupo de 3 instâncias, apenas algumas partições são movidas para a nova instância:

order-service-1  -> orders-0, orders-1        (mantém, não para)
order-service-2  -> orders-2                  (perde orders-3)
order-service-3  -> orders-4                  (perde orders-5)
order-service-4  -> orders-3, orders-5

Nota: Você não deve mudar diretamente de eager para cooperative em um group em execução. Você precisa fazer um rolling deploy em duas fases. Na primeira vez, configure ambos: CooperativeStickyAssignor,RangeAssignor. Depois, remova RangeAssignor. Se os membros não tiverem um assignor em comum, o group reportará um erro.

5.3. Static membership

A cada rolling restart, uma instância sai do group e volta. O resultado é dois rebalances por pod. Static membership resolve isso. Você atribui a cada instância um identificador fixo via group.instance.id:

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

Um consumer com group.instance.id não enviará LeaveGroup ao desligar. Se ele retornar dentro de session.timeout.ms com o mesmo id, o coordinator devolve exatamente as partições antigas e não rebalance.

Nota: HOSTNAME deve ser estável entre os restarts, por exemplo, o nome do pod em um StatefulSet no Kubernetes. Com um Deployment comum, o nome do pod muda a cada vez, então o static membership não tem efeito. Duas instâncias em execução ao mesmo tempo com o mesmo id se expulsarão do group (FencedInstanceIdException). Quando concurrency > 1, Spring Kafka adiciona automaticamente sufixos -0, -1… ao id para cada consumer filho.

A compensação dessa abordagem é que quando uma instância realmente morre, suas partições não serão processadas por ninguém até que expire session.timeout.ms.

5.4. Novo protocolo de consumer group

A partir do Kafka 4.0, o novo protocolo de rebalance (KIP-848) está pronto para produção. Você o ativa com group.protocol=consumer. A distribuição de partições é completamente transferida para o broker e sempre ocorre de forma incremental. Os consumers não precisam mais esperar uns pelos outros por uma “barreira” comum de JoinGroup.

Com este protocolo, partition.assignment.strategy no lado do cliente não é mais usado. Em vez disso, o broker escolhe o assignor (configuração no lado do cliente é group.remote.assignor). Se seu cluster e cliente forem suficientemente novos, esta é uma opção a considerar. Você deve verificar a versão kafka-clients que o Spring Boot está trazendo antes de ativar.

6. Tratamento quando partições são revogadas

Não importa quanto você otimize, o rebalance ainda acontecerá. O importante é fazer commit do offset já processado antes de perder a partição, para reduzir o processamento duplicado.

Spring Kafka já faz o commit dos offsets pendentes antes da partição ser revogada. Se você usar KafkaConsumer diretamente, você deve fazer isso por conta própria 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) {
        // Você pode recarregar state local para as novas partições aqui
    }
}

Explicação de cada parte:

• markProcessed() registra o offset próximo a ser lido, isto é, offset + 1, depois que um registro é processado.
• onPartitionsRevoked() executa na própria thread poll, logo antes da partição ser levada. Ele apenas faz commit do offset das partições que estão sendo revogadas. Este é um ponto importante com rebalancing cooperativo, pois nesse caso apenas algumas partições são revogadas.
• commitSync() bloqueia até o broker confirmar, garantindo que o offset foi salvo antes de outro consumer receber a partição.

Você registra o listener ao fazer subscribe: consumer.subscribe(List.of("orders"), listener).

Teste este comportamento usando MockConsumer fornecido em 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();
    }
}

O teste simula apenas orders-0 sendo revogado. O resultado esperado é que o offset 42 seja confirmado para orders-0. Enquanto orders-1 continua sendo mantido, então nenhum offset foi confirmado para ele.

Nota: O código em onPartitionsRevoked() deve ser rápido. Este callback executa durante o rebalance. Se for lento, todo o group tem que esperar.

7. Erros comuns

• Processamento assíncrono mas commit conforme poll. Você envia o registro para um ExecutorService, e o container faz commit do offset logo após poll(). Se um rebalance acontecer, as mensagens em processamento podem ser perdidas. Faça commit do offset apenas dos registros totalmente processados.
• Confundir session.timeout.ms com max.poll.interval.ms. Aumentar session timeout não ajuda quando a causa é processamento lento.
• Definir concurrency maior que o número de partições. Os consumers extras não têm trabalho. Eles apenas tornam o group maior e o rebalance mais demorado.
• Pausa longa de GC ou CPU throttle no container. A thread de heartbeat não consegue acompanhar, o consumer é expulso do group, mesmo que o código não seja lento. Verifique as métricas da JVM e o limite de CPU.
• Não projetar processamento idempotente. O rebalance sempre pode causar processamento duplicado em algumas mensagens. Não assuma “exactly once” se ainda não usar transaction ou não tiver um mecanismo de deduplicação, por exemplo, verificar se orderId já foi processado.

8. Conclusão

O rebalance é um mecanismo essencial para que o consumer group se dimensione automaticamente e tolere falhas. O preço é que o consumer tem que pausar temporariamente. A causa mais comum que faz o consumer “parar” continuamente é o loop poll() excedendo max.poll.interval.ms. Nesse caso, o consumer é expulso do group, o commit falha, e o rebalance se repete indefinidamente.

Para minimizar, você pode:

• Manter cada batch pequeno usando max.poll.records.
• Mudar para CooperativeStickyAssignor ou o novo protocolo de consumer group.
• Usar static membership para deployments com identificador estável.
• Fazer commit do offset no momento certo em onPartitionsRevoked().

Por fim, sempre projete processamento de mensagem idempotente. O rebalance não pode ser eliminado completamente, apenas pode ser feito para ocorrer menos frequentemente e causar menos danos.