Monitoreo de Kafka: consumer lag y métricas clave
1. Descripción general
Un sistema Kafka puede parecer “saludable”: los brokers siguen corriendo, los productores siguen enviando mensajes. Pero los consumidores podrían estar rezagados detrás de millones de mensajes sin que nadie lo sepa. La métrica que refleja este estado es el consumer lag.
En este artículo, aprenderás qué es el consumer lag y cómo medirlo de tres formas: CLI, AdminClient en Java, y Micrometer en Spring Boot. Luego, repasaremos las métricas más importantes del broker, productor y consumidor, y cómo establecer alertas para ellas.
2. ¿Qué es el consumer lag?
Cada partición en Kafka es un log de solo escritura secuencial. Hay dos offsets que debes conocer:
- Log-end offset (LEO): el offset del mensaje siguiente que se escribirá en la partición.
- Committed offset: el offset que el consumer group ha confirmado, es decir, la posición desde la que el grupo seguirá leyendo si se reinicia.
El lag de una partición se calcula así:
lag = log-end offset - committed offsetEl lag del grupo consumidor es la suma del lag de todas las particiones que el grupo lee.
Un lag de 0 significa que el consumidor está actualizado. Las fluctuaciones ligeras de lag son normales, ya que los mensajes llegan continuamente. El lag que crece consistentemente con el tiempo es la señal de alarma. En este caso, el consumidor está procesando más lentamente que el productor está escribiendo.
Nota: Con tópicos que usan transacciones o compactación, los offsets pueden tener “brechas” (registros de control, mensajes eliminados). Dado que el lag se calcula por offset, puede ser ligeramente mayor que el número de mensajes que realmente quedan por leer. Esto no es un error.
3. Verificar el lag con CLI
La forma más rápida es usar el script que viene con Kafka:
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
--describe --group order-serviceEl resultado se ve así:
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.7La partición 1 está rezagada en 3430 mensajes, mientras que las otras dos están casi al día. Un lag desparejo como este generalmente tiene dos causas. Una es que hay una clave “caliente” acumulándose en una partición. La otra es que el consumidor en el host 10.0.0.6 tiene un problema.
CLI es muy útil cuando investigas problemas. Sin embargo, no puedes correr este comando cada minuto. La siguiente sección automatizará la medición del lag.
4. Calcular el lag con AdminClient
Añade la dependencia kafka-clients (Spring Boot ya gestiona la versión):
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</dependency>Primero, extrae la lógica de cálculo en una función pura para facilitar las pruebas:
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;
}
}Esta función solo calcula el lag para particiones que tienen tanto committed offset como end offset. Math.max(0L, …) tiene su razón de ser. El committed offset y el end offset se obtienen mediante dos solicitudes diferentes, así que a veces el committed offset es “más nuevo” que el end offset que acabas de recuperar. Un resultado negativo sería sin sentido, así que lo establecemos a 0.
A continuación está la clase que llama a 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);
}
}Los pasos ocurren de la siguiente manera:
listConsumerGroupOffsetsdevuelve el committed offset del grupo. El valor puede sernullsi la partición nunca ha sido confirmada, así que lo filtramos.listOffsetsconOffsetSpec.latest()obtiene el log-end offset de cada partición.- Cada llamada tiene un timeout de 10 segundos. De esta manera, el thread no se cuelga indefinidamente si el cluster encuentra 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 -> 04.1. Pruebas para LagCalculator
Usa JUnit 5 y AssertJ (incluidos en 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. Exponer el lag en Micrometer con Spring Boot
Añade Actuator y el registro de Prometheus:
<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,prometheus5.1. Métricas del lado del cliente listas para usar
Cuando tienes MeterRegistry, Spring Boot automáticamente adjunta MicrometerConsumerListener a ConsumerFactory. Todas las métricas internas del consumidor de Kafka aparecerán en /actuator/prometheus, incluyendo:
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.0records-lag-max es el lag máximo que este consumidor ha observado. El cliente de Kafka lo calcula desde la posición de fetch, no desde el committed offset.
5.2. Gauge de lag medido desde fuera
Las métricas del lado del cliente tienen una gran debilidad. Cuando el consumidor muere, la métrica también desaparece. Justo cuando el lag está aumentando más rápido, ya no tienes datos. Por lo tanto, debes medir además el lag desde un proceso independiente 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;
}
}Recuerda habilitar scheduling con @EnableScheduling en una clase @Configuration. Cada 30 segundos, el gauge kafka_consumer_group_lag se actualiza. Este valor se basa en el committed offset, por lo que sigue siendo correcto incluso si todos los consumidores del grupo se han detenido.
En producción, muchos equipos usan herramientas dedicadas como kafka-exporter o Burrow en lugar del código anterior. El principio es el mismo: comparar committed offset con log-end offset.
Nota: No coloques este gauge en el servicio consumidor en sí si el servicio tiene múltiples instancias. Cada instancia expondrá el mismo valor de lag, y el dashboard sumará incorrectamente. Ejecútalo en un servicio de monitoreo separado, o solo en una instancia.
6. Métricas a vigilar
El lag te dice que hay un problema. Las métricas a continuación te ayudan a encontrar dónde está el problema.
6.1. Broker
| Métrica (JMX) | Significado | Valor deseado |
|---|---|---|
UnderReplicatedPartitions |
Particiones con réplicas no sincronizadas | 0 |
OfflinePartitionsCount |
Particiones sin líder | 0 |
ActiveControllerCount |
Número de controladores activos (total del cluster) | exactamente 1 |
IsrShrinksPerSec / IsrExpandsPerSec |
ISR se encoge/expande | cercano a 0, sin fluctuaciones continuas |
RequestHandlerAvgIdlePercent |
Porcentaje de inactividad del thread handler de solicitudes | > 0.3 |
NetworkProcessorAvgIdlePercent |
Porcentaje de inactividad del thread de red | > 0.3 |
OfflinePartitionsCount > 0 es un problema grave: esa partición no se puede leer ni escribir. UnderReplicatedPartitions prolongado generalmente indica que los brokers están sobrecargados, el disco es lento o hay problemas de red.
6.2. Productor
record-error-rate: tasa de envíos fallidos después de agotar los reintentos. Cualquier valor diferente de 0 merece investigación.request-latency-avg: latencia de las solicitudes al broker. Si aumenta, verás que el throughput disminuye.record-retry-rate: muchos reintentos significa que los brokers o la red no son estables.buffer-available-bytes: buffer bajo hace quesend()se bloquee.
6.3. Consumidor
records-lag-max: ya se mencionó arriba.records-consumed-rate: tasa de consumo. Compáralo con la tasa de escritura del productor en el tópico.rebalance-rate-per-hourylast-rebalance-seconds-ago: los rebalances continuos hacen que el consumidor se “congele”. Este es un culpable común del lag creciente.commit-latency-avg: los commits lentos reducen el throughput, especialmente con commits síncronos.last-poll-seconds-ago: si superamax.poll.interval.ms, el consumidor es expulsado del grupo.
7. Establecer alertas efectivas
El error más común es establecer un umbral absoluto como “alert si lag > 1000”. Para un tópico con 50.000 mensajes/segundo, 1000 es un número sin sentido. Para un tópico con unos pocos mensajes/minuto, 1000 podría significar un retraso de horas.
Hay dos enfoques mejores. El primero es alertar cuando el lag crece consistentemente:
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: "El lag de {{ $labels.group }} en {{ $labels.topic }} crece continuamente durante 15 minutos"Esta regla solo se activa cuando ambas condiciones son verdaderas: el lag tiene tendencia a crecer (deriv > 0) y ha superado el mínimo. La condición debe persistir durante 15 minutos. De esta manera, ignoras los aumentos breves después de los despliegues.
El segundo enfoque es convertir el lag a tiempo. Divide el lag por la tasa de consumo para estimar cuántos segundos atrás está el consumidor. Este número está directamente relacionado con tu SLA. Por ejemplo: “los pedidos deben procesarse en 2 minutos” es mucho más comprensible que “lag menor que 5000”.
Además, debes tener una alerta separada para cuando la métrica desaparece, por ejemplo usando absent(kafka_consumer_group_lag). La falta de datos no significa que todo esté bien.
8. Errores comunes
- Confiar solo en
records-lag-max. Cuando el consumidor muere, esta métrica desaparece. Siempre ten una fuente de medición de lag externa. - Leer el lag de un grupo sin consumidores activos. CLI aún muestra el committed offset, pero la columna
CONSUMER-IDestá vacía. El lag creciente aquí significa que nadie está procesando. - Aumentar consumidores más allá del número de particiones. Los consumidores adicionales estarán inactivos; el lag no disminuirá. Para aumentar el paralelismo, debes aumentar el número de particiones.
- Ignorar el rebalance. Procesar un lote demasiado lentamente hace que superes
max.poll.interval.ms. El consumidor es expulsado, el grupo se rebalancea, y se repite. El lag crece de manera irregular. Reducemax.poll.recordso delega el procesamiento pesado a otro thread. - Llamar a
AdminClientdemasiado frecuentemente. Consultar offsets cada segundo para cientos de grupos crea carga innecesaria en los brokers. Un intervalo de 15-60 segundos es suficiente para la mayoría de sistemas.
9. Conclusión
El consumer lag es la métrica más directa para saber si tu sistema Kafka está al día con los datos. El lag se calcula como log-end offset menos committed offset. Puedes ver el lag con kafka-consumer-groups.sh, calcularlo con AdminClient, o exportarlo a Prometheus a través de Micrometer.
Para monitorear efectivamente, combina las métricas del lado del cliente con una fuente de medición independiente. Además del lag, monitorea métricas del broker como UnderReplicatedPartitions e OfflinePartitionsCount, así como la frecuencia de rebalance en el lado del consumidor. Finalmente, establece alertas basadas en tendencias y en tiempo de retraso, en lugar de en un número de lag fijo. De esta manera, evitarás falsas alarmas sin perder problemas reales.
¿Te gusta este tipo de explicación? Cris The Dev publica un vídeo corto cada día en YouTube y TikTok.
Más sobre este tema
Consumer group y rebalance en Kafka con Spring Boot
Comprende cómo Kafka asigna particiones al consumer group, cuándo ocurre el rebalance y cómo reducir su impacto en aplicaciones Spring Boot
Change Data Capture con Debezium y Kafka
Usa Debezium para capturar cambios de PostgreSQL, envíalos a Kafka y consume con Spring Boot, eliminando el riesgo del dual-write.
Event sourcing con Kafka: guardar eventos en lugar de estado
Usa Kafka como event log para Order: registra eventos inmutables, reconstruye el estado con replay, crea read models con projection. Incluye limitaciones que debes conocer antes de aplicar.
Kafka producer: acks, idempotencia y batching
Configura acks, enable.idempotence, batch.size y linger.ms en Kafka producer para no perder mensajes, evitar duplicados y alcanzar alto throughput.
Outbox pattern con Kafka: no pierdas eventos al escribir
Escribir en base de datos y enviar a Kafka simultáneamente pierde eventos fácilmente. Este artículo enseña a implementar Outbox pattern con Spring Boot 3 y Kafka para mantener coherencia.