Grupos de consumer e rebalance no Kafka com Spring Boot

1. Visão geral

Quando você executa múltiplas instâncias do mesmo serviço lendo do Kafka, você precisa saber qual instância processa qual mensagem. O Kafka resolve isso usando consumer groups e o mecanismo de rebalance.

Este artigo explica como o Kafka distribui partições dentro de um grupo e em quais situações ocorre rebalance. Depois você configurará o Spring Boot para que rebalances ocorram com menos frequência e sem perder ou duplicar dados. No final do artigo, há um teste com Embedded Kafka para verificar como as partições são distribuídas.

2. Como os grupos de consumer funcionam

Um consumer group é um conjunto de consumers que compartilham o mesmo group.id. O Kafka garante:

Cada partição de um tópico é atribuída a exatamente um consumer do grupo em um dado momento.
Um consumer pode receber múltiplas partições.
Grupos diferentes leem de forma independente. Cada grupo tem seus próprios offsets.

Suponha que o tópico orders tenha 4 partições:

Grupo "billing" com 2 consumers:
  consumer-A -> orders-0, orders-1
  consumer-B -> orders-2, orders-3

Grupo "billing" com 5 consumers:
  consumer-A -> orders-0
  consumer-B -> orders-1
  consumer-C -> orders-2
  consumer-D -> orders-3
  consumer-E -> (ocioso, sem partição)

Uma consequência importante é que o número de partições limita o paralelismo máximo de um grupo. Adicionar consumers além do número de partições não acelera o processamento.

A coordenação é feita por um broker, chamado de group coordinator. Os consumers enviam heartbeats regularmente ao coordenador para indicar que ainda estão vivos.

2.1. Dependência

Adicione Spring for Apache Kafka. O Spring Boot 3.x já gerencia a versão:

<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. Um listener simples

@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" coloca o consumer no grupo billing.
concurrency = "2" cria 2 consumers (2 threads) na mesma instância. Se você executar 2 instâncias do serviço, o grupo terá 4 consumers, cada um recebendo 1 partição.

O log durante a inicialização mostra o resultado da distribuição:

billing: partições atribuídas: [orders-0, orders-1]
billing: partições atribuídas: [orders-2, orders-3]

3. O que é rebalance e quando ocorre

Rebalance é o processo de redistribuição de partições entre os membros do grupo. Ele ocorre quando:

Um novo consumer se junta, por exemplo quando você escalona mais pods ou implanta uma nova instância.
Um consumer sai normalmente, por exemplo um shutdown que chama close().
Um consumer é considerado morto: o coordenador não recebe heartbeat dentro de session.timeout.ms.
Um consumer processa muito lentamente: o intervalo entre duas chamadas poll() excede max.poll.interval.ms. O consumer sai automaticamente do grupo.
O tópico muda: número de partições aumenta, ou um novo tópico corresponde ao padrão de inscrição.

O problema é que no protocolo eager antigo, todos os consumers devem devolver todas as suas partições antes de serem reatribuídas. Todo o grupo para de processar neste período, um fenômeno geralmente chamado de ‘stop-the-world’. Com grupos grandes ou deployments rolling, o rebalance pode ocorrer repetidamente.

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`	Tempo máximo sem heartbeat antes do consumer ser removido
`heartbeat.interval.ms`	Intervalo de envio de heartbeat, geralmente ≤ 1/3 de `session.timeout.ms`
`max.poll.interval.ms`	Intervalo máximo entre duas chamadas `poll()`
`max-poll-records`	Número máximo de registros por chamada `poll()`

Heartbeats são enviados por uma thread de background separada. Portanto, o consumer permanece ‘vivo’ de acordo com session.timeout.ms mesmo quando a thread principal está ocupada. No entanto, max.poll.interval.ms mede apenas a thread de processamento. Se um lote leva muito tempo para ser processado, o consumer ainda é removido do grupo.

Um erro típico no log se parece com isto:

Membro consumer-billing-1 enviando solicitação LeaveGroup ao coordenador devido a
expiração do tempo limite de poll do consumer. Isso significa que o tempo entre chamadas sucessivas
a poll() foi maior que o max.poll.interval.ms configurado ...
CommitFailedException: Commit não pode ser concluído porque o grupo já fez rebalance

A fórmula que deve ser satisfeita:

max-poll-records × tempo de processamento de 1 registro  <  max.poll.interval.ms

Nota: Não ‘corrija’ aumentando max.poll.interval.ms para várias horas. Um consumer realmente travado manterá as partições por muito tempo sem que ninguém as processe. Reduza max-poll-records ou mova tarefas pesadas para fora da thread de consumo.

5. Reduzindo o impacto do rebalance

5.1. Rebalance cooperativo

Com CooperativeStickyAssignor, apenas as partições que realmente precisam ser movidas são revogadas. Os consumers restantes continuam processando suas partições durante o rebalance.

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

Exemplo: o grupo tem A (orders-0..3), depois B se junta.

Eager:       A devolve todas as 4 partições -> redistribui: A {0,1}, B {2,3}
Cooperative: A apenas devolve {2,3}, continua processando {0,1} -> B recebe {2,3}

Nota: Você não pode mudar diretamente um grupo que está em execução de eager para cooperative em um único deployment. Você precisa fazer rolling deploy duas vezes. Na primeira vez, declare ambos os assignors, por exemplo CooperativeStickyAssignor,RangeAssignor. Na segunda vez, remova RangeAssignor.

A partir do Kafka 4.0, o novo protocolo de rebalance (KIP-848) está em GA. Você o ativa com group.protocol=consumer. Neste caso, a distribuição é feita pelo broker, rebalance é incremental por padrão, e os parâmetros partition.assignment.strategy e session.timeout.ms do lado do cliente não são mais usados. Se seu cluster o suportar, esta é a opção a ser considerada em primeiro lugar.

5.2. Static membership

Cada vez que o pod reinicia, o consumer recebe um novo member.id, e isso dispara rebalance duas vezes (sai e entra novamente). Static membership atribui um identificador fixo ao consumer:

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

Se o consumer voltar dentro de session.timeout.ms, o coordenador reatribui exatamente as mesmas partições sem necessidade de rebalance. Esta abordagem é muito apropriada para Kubernetes StatefulSets, onde HOSTNAME é estável (billing-0, billing-1…).

Nota: group.instance.id deve ser único dentro do grupo. Com concurrency > 1, o Spring Kafka adiciona automaticamente um sufixo -n para cada sub-consumer. Se duas instâncias usarem o mesmo ID, a instância que entra depois ‘removera’ a anterior (FencedInstanceIdException). Não use static membership com Deployments que têm nomes de pod aleatórios, pois você não terá nenhum benefício.

6. Gerenciando offsets durante rebalance

O maior risco do rebalance é o processamento duplicado. O consumer A terminou de processar um registro mas não fez commit, a partição é transferida para B, e B lê novamente a partir do último offset confirmado.

O Spring Kafka por padrão faz commit após cada lote (AckMode.BATCH) e faz commit automaticamente dos offsets pendentes antes da revogação da partição. Se você precisar intervir, por exemplo, fazer flush de buffer ou registrar logs, 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("Prestes a perder partição {}, fazendo flush dos dados em memória", partitions);
    }

    @Override
    public void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer,
                                               Collection<TopicPartition> partitions) {
        log.info("Commit realizado e partição {} devolvida", partitions);
    }

    @Override
    public void onPartitionsAssigned(Consumer<?, ?> consumer,
                                     Collection<TopicPartition> partitions) {
        log.info("Partição {} recebida", partitions);
    }

    @Override
    public void onPartitionsLost(Consumer<?, ?> consumer,
                                 Collection<TopicPartition> partitions) {
        log.warn("Partição {} perdida sem ter feito commit", partitions);
    }
}

onPartitionsRevokedBeforeCommit é executado antes do container fazer commit do offset. Este é o lugar para finalizar o trabalho incompleto.
onPartitionsRevokedAfterCommit é executado depois que o offset foi confirmado.
onPartitionsLost é chamado quando o consumer é removido do grupo, por exemplo por exceder max.poll.interval.ms. Neste momento, a partição pode já pertencer a outro consumer, então não faça commit.

Registre o listener na 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(...) preserva toda a configuração de application.yml. A linha seguinte apenas adiciona o rebalance listener.

Nota: Não importa quão bem você configure, o Kafka apenas garante at-least-once para consumers comuns. Torne seu processamento idempotente, por exemplo verificando se orderId já foi pago antes de chamar PaymentService. Não assuma que o rebalance nunca causará duplicação.

7. Testando a distribuição de partições

O teste abaixo usa @EmbeddedKafka para verificar que dois consumers do mesmo grupo dividem igualmente 4 partições:

@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);
        }
    }
}

Explicação:

@EmbeddedKafka inicia um broker em memória e cria o tópico orders com 4 partições. O broker é injetado diretamente como parâmetro do método de teste.
Dois KafkaConsumers compartilham group.id = "billing".
O loop alterna chamadas poll() entre os dois consumers, porque rebalance ocorre apenas dentro de poll(). Se apenas o consumer A fizer poll, o rebalance não será concluído. Os dois consumers rodam na mesma thread mas são objetos separados, então não violam a regra ‘um consumer deve ser usado apenas por uma thread’.
As assertions verificam que cada consumer tem 2 partições, não se sobrepõem e totalizam as 4 partições.

Esta também é uma boa forma de testar diferentes partition.assignment.strategys. Adicione um terceiro consumer, feche um consumer e observe como assignment() muda.

8. Erros comuns

Número de consumers maior que número de partições: consumers excedentes ficarão ociosos. Aumentar concurrency não torna o sistema mais rápido.
Processamento síncrono muito lento no listener: exceder max.poll.interval.ms causa um loop de rebalance. O consumer é removido, retorna, recebe o mesmo lote novamente e é removido de novo.
Commit manual em onPartitionsLost: causa CommitFailedException, ou pior, sobrescreve o offset de um novo consumer.
Usar o mesmo group.id para dois serviços diferentes: os dois serviços ‘compartilharão’ mensagens em vez de cada um receber todas. Cada serviço independente deve ter seu próprio grupo.
Rolling deploy com protocolo eager em grupos grandes: cada restart de pod causa dois stop-the-world. Use cooperative, static membership ou o novo protocolo do Kafka 4.x.

9. Conclusão

Consumer groups permitem que você escale a leitura do Kafka horizontalmente, com limite superior sendo o número de partições. Rebalance é o preço que se paga por essa flexibilidade. Ele ocorre quando membros entram, saem, morrem ou processam muito lentamente.

Para minimizar o impacto do rebalance, você deve fazer o seguinte:

Configure adequadamente max.poll.interval.ms e max-poll-records.
Use CooperativeStickyAssignor ou group.protocol=consumer no Kafka 4.x.
Ative static membership quando a infraestrutura permitir.
Manipule o ciclo de vida das partições através de ConsumerAwareRebalanceListener.

Por fim, sempre projete seu processamento como idempotente, pois at-least-once é a garantia padrão que você deve aceitar.