ApsaraMQ for Kafka:Troubleshoot frequent consumer rebalances in ApsaraMQ for Kafka

Sintoma

Ocorrem rebalanceamentos frequentes no cliente consumidor ao usar o ApsaraMQ for Kafka.

Causa

A causa raiz depende da versão do cliente, pois o mecanismo de heartbeat difere entre as versões.

• Clientes anteriores à versão 0.10.1.0: Não há uma thread dedicado a heartbeats. O cliente envia os heartbeats dentro da chamada poll(). Quando o processamento de mensagens demora mais que session.timeout.ms, o broker não recebe o heartbeat, remove o consumidor do grupo e aciona um rebalanceamento.

• Clientes da versão 0.10.1.0 ou posterior: Uma thread dedicada a heartbeats executa de forma independente do processamento de mensagens. No entanto, se o consumidor não chamar poll() dentro do intervalo definido por max.poll.interval.ms (padrão: 5 minutos), o cliente sai do grupo de consumidores e aciona um rebalanceamento. O valor padrão de max.poll.interval.ms é 5 minutos. Isso geralmente ocorre quando o processamento de um lote de mensagens demora muito.

As seguintes mensagens de erro comumente estão associadas a rebalanceamentos. Identificar o erro específico ajuda a localizar a causa raiz mais rapidamente:

• **NOT_COORDINATOR / join group failed**: Indica que o Group Coordinator mudou (por exemplo, devido a failover do broker) ou que a expiração do tempo limite do heartbeat removeu o consumidor do grupo. O consumidor tenta reingressar no grupo, acionando um rebalanceamento.

• **MemberIdRequiredException / consumer poll timeout**: Para clientes anteriores à versão 0.10.1.0, isso geralmente ocorre devido a um consumidor travado que bloqueia o heartbeat. Para clientes da versão 0.10.1.0 ou posterior, indica que o processamento lento de mensagens fez o consumidor exceder max.poll.interval.ms sem chamar poll(), o que leva o cliente a sair do grupo de consumidores.

• **CommitFailedException**: Indica que o consumidor demorou muito para processar mensagens ou não enviou heartbeats a tempo. O broker remove o consumidor do grupo (acionando um rebalanceamento) antes de concluir o commit do offset.

Parâmetros principais

Os parâmetros a seguir controlam o comportamento do rebalanceamento:

Soluções

1. Ajuste os valores dos parâmetros/@cmd

   Defina os seguintes parâmetros com base na versão do cliente:

   **session.timeout.ms**

   Versão do cliente	Valor recomendado
   Anterior a 0.10.1.0	Maior que o tempo de processamento de um lote de mensagens, mas não superior a 30 segundos. Recomenda-se 25 segundos.
   0.10.1.0 ou posterior	Mantenha o valor padrão de 10 segundos.

   **max.poll.records**

   Defina este valor bem abaixo do resultado da seguinte fórmula:

   max.poll.records << messages_per_thread_per_second * number_of_threads * max.poll.interval.ms

   **max.poll.interval.ms** (apenas para a versão 0.10.1.0 ou posterior)

   Defina este valor acima do resultado da seguinte fórmula:

   max.poll.interval.ms > max.poll.records / (messages_per_thread_per_second * number_of_threads)

   **heartbeat.interval.ms** (versão 0.10.1.0 ou posterior)

   Defina heartbeat.interval.ms como no máximo 10.000 milissegundos (10 segundos). Esse valor controla a frequência com que o consumidor envia heartbeats ao broker e deve ser sempre menor que session.timeout.ms. Uma prática comum é defini-lo como um terço de session.timeout.ms.

   Configurar parâmetros no Spring Kafka

   Se usar o Spring Kafka, configure todos os parâmetros acima no arquivo application.yml:

   spring:
     kafka:
       consumer:
         properties:
           session.timeout.ms: 10000
           heartbeat.interval.ms: 3000
           max.poll.interval.ms: 300000
           max.poll.records: 50

   Ajuste os valores para corresponder ao throughput de consumo. O valor de max.poll.interval.ms deve exceder o tempo necessário para processar um único lote de mensagens de max.poll.records.

2. Melhore a velocidade de consumo e separe as threads de processamento/@cmd

   Transfira o processamento de mensagens para uma thread dedicado para que a thread do consumidor possa chamar poll() no intervalo programado. Isso evita que o processamento lento bloqueie os heartbeats ou exceda o intervalo de poll.

3. Reduza os tópicos por grupo de consumidores/@cmd

   Mantenha cada grupo de consumidores inscrito em no máximo cinco tópicos. Para uma estabilidade ideal, inscreva apenas um tópico por grupo de consumidores.

4. Atualize para a versão 0.10.1.0 ou posterior/@cmd

   Clientes anteriores à versão 0.10.1.0 não possuem uma thread dedicada a heartbeats, o que os torna vulneráveis à expiração do tempo limite de heartbeat durante processamentos pesados. A atualização para a versão 0.10.1.0 ou posterior desacopla o envio de heartbeats do processamento de mensagens e elimina esse tipo de rebalanceamento.

Perguntas frequentes

P: Como reduzo temporariamente os rebalanceamentos para acelerar o consumo de mensagens acumuladas?

Execute as etapas a seguir para minimizar os rebalanceamentos e limpar o acúmulo de mensagens o mais rápido possível:

1. Use a versão do cliente 0.10.1.0 ou posterior. Clientes anteriores à versão 0.10.1.0 não possuem uma thread dedicada a heartbeats e estão propensos a rebalanceamentos sob carga pesada. Atualize antes de ajustar outros parâmetros.

2. **Aumente max.poll.interval.ms** para um valor maior que o tempo máximo necessário para processar um único lote de mensagens. Isso evita que o broker considere o consumidor inativo durante um processamento pesado.

3. **Diminua max.poll.records** para reduzir o número de mensagens buscadas por chamada poll(). Lotes menores terminam mais rápido, reduzindo o risco de exceder max.poll.interval.ms.

4. Reduza o número de tópicos inscritos por grupo de consumidores. Mantenha as inscrições em no máximo cinco tópicos. Para uma estabilidade ideal, inscreva apenas um tópico por grupo de consumidores.

5. Evite bloquear a thread do consumidor. Mova a lógica de processamento de mensagens para uma thread assíncrona separada, para que a thread do consumidor permaneça livre para chamar poll() no intervalo programado.