Solucionando Atraso Comum do Consumidor Kafka Usando Comandos de Console

O atraso do consumidor é o primeiro número que a maioria dos operadores Kafka verifica quando um pipeline parece lento, mas também é um dos números mais fáceis de interpretar erroneamente. Um grupo pode mostrar um milhão de registros de atraso porque uma API downstream está com tempo limite, porque uma implantação deixou metade dos consumidores offline, porque uma partição está mais quente que as outras, ou porque o aplicativo está saudável e simplesmente alcançando após uma pausa planejada. Os comandos são simples. O julgamento em torno deles é onde os incidentes são ganhos ou perdidos.

Este guia se concentra no caminho da linha de comando que uso durante um incidente de atraso: descrever o grupo, comparar partições, confirmar se os consumidores estão ativos, decidir se o atraso está crescendo ou diminuindo, e só então considerar uma redefinição de offset. As redefinições de offset estão incluídas porque às vezes são necessárias, mas não são uma cura para um consumidor lento. Elas pulam trabalho ou reproduzem trabalho. Trate-as como uma decisão operacional, não uma correção de desempenho.

Entendendo o Atraso do Consumidor Kafka

No Kafka, as mensagens são organizadas em tópicos, que são divididos em partições. Cada mensagem dentro de uma partição recebe um offset sequencial e imutável. Os consumidores leem mensagens de uma partição mantendo sua posição atual, também conhecida como seu offset confirmado. O broker Kafka rastreia o log-end-offset para cada partição, que representa o offset da mensagem mais recente anexada a ela.

Atraso do Consumidor = Log-End-Offset - Offset Confirmado

Essencialmente, o atraso é o número de mensagens que um consumidor está atrás do cabeçalho do log para uma determinada partição. Embora algum atraso seja natural e esperado em qualquer sistema de streaming, um atraso consistentemente crescente ou excessivamente grande sinaliza um problema.

Por que o Alto Atraso do Consumidor é uma Preocupação:

• Processamento de Dados Atrasado: Seus aplicativos podem estar processando dados muito lentamente, impactando análises em tempo real ou operações críticas de negócios.
• Exaustão de Recursos: Os consumidores podem estar lutando para acompanhar, levando a alto uso de CPU, memória ou rede.
• Dados Desatualizados: Sistemas downstream que recebem dados de consumidores atrasados operarão com informações desatualizadas.
• Problemas de Política de Retenção: Se o atraso exceder o período de retenção do tópico, os consumidores podem perder mensagens permanentemente à medida que são removidas do log.
• Reequilíbrios do Grupo de Consumidores: O atraso persistente pode contribuir para um comportamento instável do grupo de consumidores e reequilíbrios frequentes.

Causas Comuns de Alto Atraso:

• Lógica Lenta do Consumidor: O próprio aplicativo consumidor está demorando muito para processar cada mensagem.
• Instâncias de Consumidor Insuficientes: Não há instâncias de consumidor suficientes em execução para lidar com o volume de mensagens em todas as partições.
• Latência de Rede: Problemas entre consumidores e brokers.
• Problemas de Desempenho do Broker: Os brokers podem estar lutando para servir mensagens de forma eficiente.
• Picos na Produção de Mensagens: Rajadas temporárias de mensagens que sobrecarregam os consumidores.
• Erros de Configuração: Configurações incorretas do consumidor ou do tópico.

Diagnosticando Atraso com kafka-consumer-groups.sh (Recomendado)

A ferramenta kafka-consumer-groups.sh é a maneira moderna e recomendada de gerenciar e inspecionar grupos de consumidores. Ela interage diretamente com os brokers Kafka para recuperar informações de offset do consumidor, que são armazenadas em um tópico interno __consumer_offsets. Esta ferramenta fornece detalhes abrangentes sobre o estado do grupo de consumidores, incluindo atraso.

Uso Básico para Descrever um Grupo de Consumidores

Para verificar o atraso de um grupo de consumidores específico, use as opções --describe e --group:

kafka-consumer-groups.sh --bootstrap-server <Kafka_Broker_Host:Port> --describe --group <Consumer_Group_Name>

Substitua <Kafka_Broker_Host:Port> pelo endereço de um dos seus brokers Kafka (por exemplo, localhost:9092) e <Consumer_Group_Name> pelo nome do grupo de consumidores que você deseja inspecionar.

Interpretando a Saída

Uma saída típica será mais ou menos assim:

GROUP           TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                       HOST            CLIENT-ID
my-consumer-app my-topic                       0          12345           12347           2               consumer-1-a1b2c3d4-e5f6-7890-1234-abcdedfg      /192.168.1.100  consumer-1
my-consumer-app my-topic                       1          20000           20500           500             consumer-2-hijk-lmno-pqrs-tuvw-xyz              /192.168.1.101  consumer-2
my-consumer-app my-topic                       2          5000            5000            0               consumer-3-1234-5678-90ab-cdef-12345678          /192.168.1.102  consumer-3
my-consumer-app another-topic                  0          900             900             0               consumer-1-a1b2c3d4-e5f6-7890-1234-abcdedfg      /192.168.1.100  consumer-1

Vamos detalhar as colunas importantes:

• GROUP: O nome do grupo de consumidores.
• TOPIC: O tópico que está sendo consumido.
• PARTITION: A partição específica do tópico.
• CURRENT-OFFSET: O último offset confirmado pelo consumidor para esta partição.
• LOG-END-OFFSET: O offset da mensagem mais recente nesta partição.
• LAG: A diferença entre LOG-END-OFFSET e CURRENT-OFFSET. Este é o número de mensagens que o consumidor está atrasado.
• CONSUMER-ID: Um identificador único para a instância do consumidor. Se for -, significa que nenhum consumidor ativo está atribuído a essa partição.
• HOST: O endereço IP ou nome do host da instância do consumidor.
• CLIENT-ID: O ID do cliente configurado para a instância do consumidor.

Observações Principais:

• Valores altos de LAG: Indicam que o consumidor está ficando para trás. Investigue a lógica do consumidor, recursos ou escalonamento.
• - em CONSUMER-ID: Sugere que uma partição não está sendo consumida. Isso pode ser devido a um número insuficiente de consumidores ativos no grupo ou a uma instância do consumidor que travou sem se juntar novamente. Se LAG for alto para tais partições, é um problema crítico.
• LAG de 0: Significa que o consumidor está totalmente atualizado com as mensagens mais recentes.

Diagnosticando Atraso com consumer-offset-checker.sh (Ferramenta Legada)

consumer-offset-checker.sh é uma ferramenta mais antiga e obsoleta que dependia do ZooKeeper para armazenar e recuperar offsets do consumidor (para consumidores que usam o antigo kafka.consumer.ZookeeperConsumerConnector). Para clientes Kafka modernos (0.9.0 e posteriores), os offsets são armazenados no próprio Kafka. Embora tenha sido amplamente substituída por kafka-consumer-groups.sh, você pode encontrá-la em ambientes mais antigos ou com clientes consumidores legados.

Aviso: Aviso de Descontinuação

Esta ferramenta depende do ZooKeeper para gerenciamento de offset. Clientes Kafka modernos (0.9.0+) armazenam offsets diretamente no Kafka. Para clusters e clientes mais novos, kafka-consumer-groups.sh é a ferramenta autoritativa e preferida. Use consumer-offset-checker.sh apenas se você souber explicitamente que seus clientes consumidores estão configurados para armazenar offsets no ZooKeeper.

Uso Básico

Para verificar o atraso com esta ferramenta, você precisa fornecer a string de conexão do ZooKeeper:

consumer-offset-checker.sh --zk <ZooKeeper_Host:Port> --group <Consumer_Group_Name>

Substitua <ZooKeeper_Host:Port> (por exemplo, localhost:2181) e <Consumer_Group_Name>.

Interpretando a Saída

Group           Topic                          Partition Offset  LogSize Lag     Owner
my-old-app      my-old-topic                   0         1000    1050    50      consumer-1_hostname-1234-5678-90ab-cdef
my-old-app      my-old-topic                   1         2000    2000    0       consumer-2_hostname-abcd-efgh-ijkl-mnop

• Group, Topic, Partition: Semelhante a kafka-consumer-groups.sh.
• Offset: O offset confirmado pelo consumidor.
• LogSize: O LOG-END-OFFSET da partição.
• Lag: O número de mensagens que o consumidor está atrasado.
• Owner: A instância do consumidor atualmente proprietária (consumindo) da partição.

A interpretação dos valores de atraso é semelhante: alto atraso indica problemas, e um Owner ausente para uma partição de alto atraso é um problema crítico.

Lidando com Alto Atraso do Consumidor: Estratégias e Redefinições de Offset

Depois de identificar o alto atraso do consumidor, o próximo passo é lidar com ele. Isso geralmente envolve uma abordagem dupla: primeiro, investigar e corrigir a causa raiz e, segundo, se necessário, redefinir os offsets do consumidor.

Investigando a Causa Raiz

Antes de pular para as redefinições de offset, é crucial entender por que o atraso está ocorrendo. Verifique o seguinte:

• Logs do Aplicativo Consumidor: Procure erros, tempos de processamento excessivos ou sinais de falha do aplicativo.
• Métricas do Host do Consumidor: Monitore o uso de CPU, memória e rede. O consumidor está com recursos limitados?
• Métricas do Broker Kafka: Os brokers estão sob estresse? O disco I/O, rede ou CPU estão altos?
• Taxa de Transferência do Produtor: Houve um pico inesperado na produção de mensagens?
• Estado do Grupo de Consumidores: Há reequilíbrios frequentes? O max.poll.interval.ms está sendo atingido?

Escalonando Consumidores

Se o problema é que os consumidores existentes não conseguem processar mensagens rápido o suficiente e o tópico tem partições suficientes, você pode precisar escalar seu grupo de consumidores adicionando mais instâncias de consumidor. Cada instância do consumidor em um grupo assumirá uma ou mais partições até que todas as partições sejam atribuídas, até o número de partições.

Redefinindo Offsets do Consumidor

Redefinir offsets do consumidor significa alterar o ponto de partida a partir do qual um grupo de consumidores lerá as mensagens. Esta é uma operação poderosa e potencialmente disruptiva que deve ser usada com cautela.

Considerações Importantes Antes de Redefinir Offsets:

• Perda de Dados: Redefinir para --to-latest fará com que os consumidores pulem todas as mensagens entre seu offset atual e o log-end-offset, levando à perda permanente de dados para essas mensagens.
• Reprocessamento de Dados: Redefinir para --to-earliest ou um offset mais antigo significa que os consumidores reprocessarão mensagens que já manipularam. Seu aplicativo consumidor deve ser idempotente (processar uma mensagem várias vezes produz o mesmo resultado) para lidar com isso de forma adequada.
• Estado do Aplicativo: Considere como o reprocessamento pode afetar qualquer estado gerenciado por seu aplicativo consumidor ou sistemas downstream.

Para redefinir offsets, você usará novamente kafka-consumer-groups.sh. Ele oferece várias opções para como redefinir offsets:

• --to-earliest: Redefine os offsets para o offset mais antigo disponível na partição.
• --to-latest: Redefine os offsets para o offset mais recente na partição (efetivamente pulando todas as mensagens atuais).
• --to-offset <offset>: Redefine os offsets para um offset específico e desejado.
• --to-datetime <YYYY-MM-DDTHH:mm:SS.sss>: Redefine os offsets para o offset correspondente a um timestamp específico.
• --shift-by <N>: Desloca o offset atual em N posições (por exemplo, -10 para voltar 10 mensagens, +10 para avançar 10 mensagens).

Recursos de Segurança Cruciais: --dry-run e --execute

Sempre execute um --dry-run primeiro para ver o que a operação de redefinição faria antes de confirmar com --execute.

Processo Passo a Passo para Redefinir Offsets:

1. Pare todos os consumidores no grupo de consumidores alvo. Isso é vital para evitar que os consumidores confirmem novos offsets enquanto você tenta redefini-los.

2. Execute uma simulação (dry run) para visualizar as alterações de offset:

   • Exemplo: Redefinindo para o offset mais antigo (reprocessar todas as mensagens)

     kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-consumer-app --reset-offsets --to-earliest --topic my-topic --dry-run
     

   • Exemplo: Redefinindo para o offset mais recente (pular todas as mensagens atrasadas)

     kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-consumer-app --reset-offsets --to-latest --topic my-topic --dry-run
     

   • Exemplo: Redefinindo para um timestamp específico (por exemplo, começar de 2023-01-01 00:00:00 UTC)

     kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-consumer-app --reset-offsets --to-datetime 2023-01-01T00:00:00.000 --topic my-topic --dry-run
     

   • Exemplo: Deslocando offsets para trás em 500 mensagens (por partição)

     kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-consumer-app --reset-offsets --shift-by -500 --topic my-topic --dry-run
     

   A saída de --dry-run mostrará as alterações de offset propostas:

   GROUP                          TOPIC                          PARTITION NEW-OFFSET
   my-consumer-app                my-topic                       0         0
   my-consumer-app                my-topic                       1         0
   

3. Execute a redefinição assim que estiver satisfeito com os resultados da simulação:

   • Exemplo: Redefinindo para o offset mais antigo (executar)

     kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-consumer-app --reset-offsets --to-earliest --topic my-topic --execute
     

4. Reinicie os aplicativos consumidores. Após a redefinição dos offsets, reinicie suas instâncias de consumidor. Elas começarão a consumir a partir dos novos offsets iniciais.

Dica: Redefinindo para Todos os Tópicos em um Grupo

Se você quiser redefinir offsets para todos os tópicos consumidos por um grupo, pode omitir a flag --topic ao usar kafka-consumer-groups.sh --reset-offsets. Seja extremamente cauteloso com isso, pois afeta tudo.

Melhores Práticas para Operações do Consumidor

• Monitoramento Proativo: Implemente monitoramento robusto para atraso do consumidor usando ferramentas como Prometheus/Grafana, Datadog ou scripts personalizados. Configure alertas para atraso que cresce rapidamente ou consistentemente alto.
• Entenda a Idempotência: Projete seus aplicativos consumidores para serem idempotentes. Isso permite o reprocessamento seguro de mensagens em caso de falhas ou redefinições de offset.
• Ajuste max.poll.interval.ms: Esta configuração define o tempo máximo que um consumidor pode ficar sem fazer polling. Se sua lógica de processamento for lenta, aumente esse valor para evitar reequilíbrios indesejados, mas também investigue a lentidão subjacente.
• Lide com Mensagens Não Processáveis: Implemente uma estratégia para mensagens "poison pill" (por exemplo, enviá-las para uma Fila de Cartas Mortas - DLQ) em vez de falhar repetidamente e bloquear o consumidor.
• Desligamentos Graciosos: Certifique-se de que seus aplicativos consumidores sejam desligados corretamente, confirmando seus offsets finais para evitar reprocessamento desnecessário ou picos de atraso durante as reinicializações.
• Combine Partições com Consumidores: Para paralelismo ideal, procure ter pelo menos tantas partições quanto você espera executar instâncias de consumidor. Mais partições permitem mais paralelismo.

Um Fluxo de Incidente Prático

Quando o atraso dispara um alerta, resista ao impulso de redefinir offsets primeiro. Comece capturando o estado atual do grupo:

kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 --describe --group payments-writer

Procure pela forma, não apenas pelo tamanho. Se cada partição está atrasada aproximadamente na mesma quantidade, todo o grupo provavelmente está subdimensionado ou bloqueado em uma dependência compartilhada. Se uma partição está muito atrás, verifique se há distorção de chave, uma mensagem venenosa ou um único host consumidor com mau comportamento de CPU, disco, DNS ou rede. Se CONSUMER-ID for -, a partição não tem membro ativo atribuído naquele momento; isso geralmente aponta para consumidores travados, um reequilíbrio em andamento ou um grupo com menos membros saudáveis do que o esperado.

Execute o comando novamente um minuto depois. Um valor de atraso de 500.000 é menos preocupante se estiver caindo rapidamente após um rollback de implantação. Um valor de atraso de 5.000 é mais preocupante se dobrar a cada minuto durante o tráfego normal. Durante um incidente, geralmente anoto três números: atraso total, atraso da pior partição e se o estado do grupo está estável. Isso fornece sinal suficiente para decidir se deve escalar consumidores, desacelerar produtores, corrigir erros de aplicativo ou preparar uma reprodução controlada.

Antes de qualquer redefinição, salve os offsets atuais em algum lugar durável, mesmo que seja apenas no ticket do incidente. Uma simulação (dry run) não é um backup. A saída do comando fornece os offsets que você pode precisar se alguém perceber que a redefinição pulou dados que ainda importavam.

Verificações Finais

Um runbook de atraso saudável tem três hábitos: descrever antes de alterar, simular antes de executar e corrigir o consumidor antes de mover offsets. kafka-consumer-groups.sh fornece a verdade bruta sobre offsets confirmados e propriedade de partição. Seu trabalho é conectar essa saída ao comportamento do aplicativo por trás dela.