Monitorando Kafka: consumer lag e métricas essenciais

1. Visão geral

Um sistema Kafka pode parecer “saudável”: brokers continuam rodando, producers conseguem enviar mensagens. Mas o consumer pode estar ficando para trás por milhões de mensagens sem que ninguém saiba. A métrica que reflete essa situação é o consumer lag.

Neste artigo, você vai entender como o consumer lag é calculado e como medi-lo de três formas: CLI, AdminClient em Java e Micrometer no Spring Boot. Depois, vamos passar pelas métricas mais importantes de broker, producer e consumer, além de como configurar alertas para elas.

2. O que é consumer lag?

Cada partition no Kafka é um log somente para append. Há dois offsets que você precisa entender:

• Log-end offset (LEO): offset da mensagem seguinte que será escrita na partition.
• Committed offset: offset que o consumer group fez commit, ou seja, a posição de onde o grupo continuará lendo se reiniciar.

O lag de uma partition é calculado assim:

lag = log-end offset - committed offset

O lag de todo o consumer group é a soma do lag de todas as partitions que o grupo lê.

Lag igual a 0 significa que o consumer está em dia. Variações pequenas são normais, já que mensagens chegam continuamente. Lag aumentando consistentemente ao longo do tempo é o que deve preocupar. Nesse caso, o consumer está processando mais lentamente do que o producer está escrevendo.

Nota: Em tópicos com transações ou compaction, offsets podem ter “lacunas” (control records, mensagens deletadas). Como o lag é calculado por offset, pode ser um pouco maior que o número real de mensagens ainda por ler. Isso não é um erro.

3. Verificar lag com CLI

A forma mais rápida é usar o script que vem com Kafka:

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group order-service

O resultado tem este formato:

GROUP          TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG   CONSUMER-ID       HOST
order-service  orders  0          15230           15232           2     consumer-1-a1b2   /10.0.0.5
order-service  orders  1          14980           18410           3430  consumer-1-c3d4   /10.0.0.6
order-service  orders  2          15102           15102           0     consumer-1-e5f6   /10.0.0.7

A partition 1 está atrasada por 3430 mensagens, enquanto as outras duas estão praticamente em dia. Essas diferenças de lag geralmente têm duas causas. Uma é ter uma chave “quente” concentrada em uma partition. Outra é o consumer no host 10.0.0.6 estar com problemas.

CLI é muito útil quando investigando problemas. No entanto, você não pode ficar executando esse comando a cada minuto. A próxima seção vai automatizar a medição de lag.

4. Calcular lag com AdminClient

Adicione a dependência kafka-clients (Spring Boot já gerencia a versão):

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
</dependency>

Primeiro, separe a lógica de cálculo em uma função pura para facilitar testes:

public final class LagCalculator {

    private LagCalculator() {}

    public static Map<TopicPartition, Long> compute(Map<TopicPartition, Long> committed,
                                                    Map<TopicPartition, Long> endOffsets) {
        Map<TopicPartition, Long> lag = new HashMap<>();
        committed.forEach((tp, committedOffset) -> {
            Long end = endOffsets.get(tp);
            if (end != null) {
                lag.put(tp, Math.max(0L, end - committedOffset));
            }
        });
        return lag;
    }
}

Essa função calcula lag apenas para partitions que têm tanto committed offset quanto end offset. Math.max(0L, …) tem uma razão. Como committed offset e end offset são obtidos em duas requisições diferentes, às vezes o committed offset é “mais novo” que o end offset que foi buscado. Um resultado negativo seria sem sentido, então fixamos em 0.

Em seguida, a classe que chama AdminClient:

public class ConsumerLagChecker {

    private final Admin admin;

    public ConsumerLagChecker(Admin admin) {
        this.admin = admin;
    }

    public Map<TopicPartition, Long> lagOf(String groupId) throws Exception {
        Map<TopicPartition, Long> committed = admin.listConsumerGroupOffsets(groupId)
                .partitionsToOffsetAndMetadata()
                .get(10, TimeUnit.SECONDS)
                .entrySet().stream()
                .filter(e -> e.getValue() != null)
                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().offset()));

        Map<TopicPartition, OffsetSpec> request = committed.keySet().stream()
                .collect(Collectors.toMap(tp -> tp, tp -> OffsetSpec.latest()));

        Map<TopicPartition, Long> endOffsets = admin.listOffsets(request)
                .all()
                .get(10, TimeUnit.SECONDS)
                .entrySet().stream()
                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().offset()));

        return LagCalculator.compute(committed, endOffsets);
    }
}

Cada etapa funciona assim:

• listConsumerGroupOffsets retorna o committed offset do grupo. O valor pode ser null se a partition nunca foi commitada, então filtramos.
• listOffsets com OffsetSpec.latest() busca o log-end offset de cada partition.
• Cada chamada tem um timeout de 10 segundos. Assim, a thread não fica travada indefinidamente se o cluster tiver problemas.

Uso:

try (Admin admin = Admin.create(Map.of(
        AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"))) {
    new ConsumerLagChecker(admin).lagOf("order-service")
            .forEach((tp, lag) -> System.out.println(tp + " -> " + lag));
}

orders-0 -> 2
orders-1 -> 3430
orders-2 -> 0

4.1. Testes para LagCalculator

Use JUnit 5 e AssertJ (inclusos em spring-boot-starter-test):

class LagCalculatorTest {

    private final TopicPartition p0 = new TopicPartition("orders", 0);
    private final TopicPartition p1 = new TopicPartition("orders", 1);

    @Test
    void shouldComputeLagPerPartition() {
        Map<TopicPartition, Long> lag = LagCalculator.compute(
                Map.of(p0, 100L, p1, 50L),
                Map.of(p0, 120L, p1, 50L));

        assertThat(lag).containsEntry(p0, 20L).containsEntry(p1, 0L);
    }

    @Test
    void shouldClampNegativeLagToZero() {
        Map<TopicPartition, Long> lag = LagCalculator.compute(
                Map.of(p0, 130L),
                Map.of(p0, 120L));

        assertThat(lag).containsEntry(p0, 0L);
    }

    @Test
    void shouldSkipPartitionWithoutEndOffset() {
        Map<TopicPartition, Long> lag = LagCalculator.compute(
                Map.of(p0, 10L), Map.of());

        assertThat(lag).isEmpty();
    }
}

5. Exportando lag para Micrometer no Spring Boot

Adicione Actuator e Prometheus registry:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-registry-prometheus</artifactId>
</dependency>

management.endpoints.web.exposure.include=health,prometheus

5.1. Métricas nativas do lado do cliente

Quando você tem MeterRegistry, Spring Boot automaticamente conecta MicrometerConsumerListener ao ConsumerFactory. Todas as métricas internas do Kafka consumer aparecerão em /actuator/prometheus, incluindo:

kafka_consumer_fetch_manager_records_lag_max{client_id="consumer-order-service-1",...} 3430.0
kafka_consumer_fetch_manager_records_lag{topic="orders",partition="1",...} 3430.0

records-lag-max é o lag máximo que esse consumer particular observou. O cliente Kafka calcula automaticamente a partir da posição de fetch, não do committed offset.

5.2. Gauge de lag medido externamente

As métricas do lado do cliente têm uma grande limitação. Quando o consumer morre, a métrica também desaparece. Justamente quando o lag está crescendo mais rapidamente você perde os dados. Por isso, você deve medir lag adicionalmente de um processo independente usando AdminClient:

@Component
public class ConsumerLagMetrics {

    private final ConsumerLagChecker checker;
    private final MeterRegistry registry;
    private final Map<TopicPartition, AtomicLong> gauges = new ConcurrentHashMap<>();

    public ConsumerLagMetrics(KafkaAdmin kafkaAdmin, MeterRegistry registry) {
        this.checker = new ConsumerLagChecker(Admin.create(kafkaAdmin.getConfigurationProperties()));
        this.registry = registry;
    }

    @Scheduled(fixedDelay = 30_000)
    public void refresh() throws Exception {
        checker.lagOf("order-service").forEach((tp, lag) ->
                gauges.computeIfAbsent(tp, this::register).set(lag));
    }

    private AtomicLong register(TopicPartition tp) {
        AtomicLong holder = new AtomicLong();
        Gauge.builder("kafka.consumer.group.lag", holder, AtomicLong::get)
                .tag("group", "order-service")
                .tag("topic", tp.topic())
                .tag("partition", String.valueOf(tp.partition()))
                .register(registry);
        return holder;
    }
}

Lembre-se de habilitar agendamento com @EnableScheduling em uma classe @Configuration. A cada 30 segundos, o gauge kafka_consumer_group_lag é atualizado. Esse valor é baseado no committed offset, então permanece preciso mesmo quando todos os consumers do grupo param.

Em produção, muitas equipes usam ferramentas especializadas como kafka-exporter ou Burrow em vez do código acima. Seus princípios são os mesmos: comparar committed offset com log-end offset.

Nota: Não coloque esse gauge no próprio serviço consumer se o serviço tiver múltiplas instâncias. Cada instância exportará o mesmo valor de lag e o dashboard vai somar tudo, dando o resultado errado. Execute em um serviço de monitoramento separado ou apenas em uma única instância.

6. Métricas para monitorar

Lag te diz se há um problema. As métricas abaixo ajudam você a descobrir onde está o problema.

6.1. Broker

Métrica (JMX)	Significado	Valor esperado
UnderReplicatedPartitions	Partitions com réplicas desincronizadas	0
OfflinePartitionsCount	Partitions sem leader	0
ActiveControllerCount	Número de controllers ativos (total do cluster)	exatamente 1
IsrShrinksPerSec / IsrExpandsPerSec	ISR encolhendo/expandindo	perto de 0, sem flutuações contínuas
RequestHandlerAvgIdlePercent	Ociosidade das threads de manipulação de requisições	> 0.3
NetworkProcessorAvgIdlePercent	Ociosidade das threads de processamento de rede	> 0.3

OfflinePartitionsCount > 0 é um problema sério: aquela partition não pode ler ou escrever. UnderReplicatedPartitions prolongado geralmente indica broker sobrecarregado, disco lento ou problemas de rede.

6.2. Producer

• record-error-rate: taxa de falhas de envio após esgotar as tentativas. Qualquer valor diferente de 0 merece investigação.
• request-latency-avg: latência de requisições ao broker. Se aumentar, você verá throughput diminuir.
• record-retry-rate: muitas tentativas significam que broker ou rede estão instáveis.
• buffer-available-bytes: buffer quase cheio causa bloqueio em send().

6.3. Consumer

• records-lag-max: já mencionado acima.
• records-consumed-rate: taxa de consumo. Compare com a taxa de escrita do producer no tópico.
• rebalance-rate-per-hour e last-rebalance-seconds-ago: rebalances contínuos deixam o consumer “congelado”. Esse é um culpado comum para lag crescente.
• commit-latency-avg: commits lentos reduzem throughput, especialmente em commits síncronos.
• last-poll-seconds-ago: se exceder max.poll.interval.ms, o consumer é removido do grupo.

7. Configurando alertas apropriados

O erro mais comum é definir um limiar absoluto como “lag > 1000 então alerta”. Para um tópico com 50.000 mensagens/segundo, 1000 é um número sem significado. Para um tópico com poucas mensagens/minuto, 1000 já representa horas de atraso.

Há duas abordagens melhores. A primeira é alertar quando lag está aumentando continuamente:

groups:
  - name: kafka-consumer
    rules:
      - alert: KafkaConsumerLagGrowing
        expr: sum by (group, topic) (deriv(kafka_consumer_group_lag[10m])) > 0
          and sum by (group, topic) (kafka_consumer_group_lag) > 500
        for: 15m
        labels:
          severity: warning
        annotations:
          summary: "Lag do {{ $labels.group }} no {{ $labels.topic }} está aumentando continuamente há 15 minutos"

Essa regra só ativa quando ambas as condições são verdadeiras: lag tem tendência crescente (deriv > 0) e já ultrapassou um nível mínimo. A condição deve durar 15 minutos. Assim você evita falsos positivos por picos curtos após deploy.

A segunda é converter lag em tempo. Divida lag pela taxa de consumo para estimar quantos segundos o consumer está atrasado. Esse número se relaciona diretamente com SLA. Por exemplo: “pedidos devem ser processados em 2 minutos” é muito mais claro que “lag abaixo de 5000”.

Além disso, você deve ter um alerta separado para quando a métrica desaparece, por exemplo usando absent(kafka_consumer_group_lag). Falta de dados não significa que tudo está bem.

8. Erros comuns

• Confiando apenas em records-lag-max. Quando o consumer morre, essa métrica desaparece. Sempre tenha uma fonte externa de medição de lag.
• Lendo lag de um grupo sem consumers ativos. CLI ainda mostra committed offset, mas a coluna CONSUMER-ID fica vazia. Lag crescente aqui significa que ninguém está processando.
• Adicionando consumers além do número de partitions. Consumers extras ficarão ociosos e o lag não diminuirá. Para aumentar paralelismo, você deve aumentar o número de partitions.
• Ignorando rebalance. Processar um lote por muito tempo ultrapassa max.poll.interval.ms. O consumer é removido, o grupo rebalanceia, repete. Lag cresce em padrão de serra. Reduza max.poll.records ou mova processamento pesado para outra thread.
• Chamando AdminClient muito frequentemente. Consultar offsets a cada segundo para centenas de grupos cria carga desnecessária no broker. Um ciclo de 15–60 segundos é suficiente para a maioria dos sistemas.

9. Conclusão

Consumer lag é a métrica mais direta para saber se seu sistema Kafka está acompanhando os dados. Lag é calculado como log-end offset menos committed offset. Você pode ver lag com kafka-consumer-groups.sh, calculá-lo com AdminClient ou exportá-lo para Prometheus via Micrometer.

Para monitorar efetivamente, combine métricas do lado do cliente com uma fonte de medição independente. Junto ao lag, acompanhe métricas de broker como UnderReplicatedPartitions e OfflinePartitionsCount, além da frequência de rebalance no lado do consumer. Finalmente, alerte baseado em tendências e tempo de latência, não em um número fixo de lag. Dessa forma você terá menos falsos alarmes e não perderá problemas reais.