¿Qué causa el consumer lag en una cola de mensajes, y cómo manejas el backpressure cuando los consumidores no dan abasto?

El consumer lag es la distancia, medida en offsets, entre el último mensaje escrito en una partición (el log end offset) y el último offset que un consumer group realmente ha confirmado como procesado. Un lag de cero significa que el consumidor está al día en tiempo real; un lag de 50.000 significa que hay 50.000 mensajes sin procesar esperando en esa partición a ese consumer group. El lag se rastrea por partición, y un promedio que se ve saludable puede esconder una sola partición gravemente atascada — que es justo el modo de fallo que a los entrevistadores les gusta explorar.

#Qué causa realmente el lag

El lag crece siempre que la tasa de procesamiento de un consumer group cae por debajo de la tasa de producción del topic, por alguna de varias razones distintas:

Una clave de partición desbalanceada empeora varias de estas causas: si una sola clave (digamos, un customerId muy activo) concentra una porción desproporcionada del tráfico, su partición puede acumular un lag severo mientras todas las demás se ven perfectamente sanas.

#Monitorear el lag

Kafka expone directamente los offsets confirmados y los log-end-offsets, así que el lag se puede calcular sin infraestructura adicional. La forma más rápida de revisarlo es la CLI:

kafka-consumer-groups.sh --bootstrap-server broker1:9092 \
  --describe --group billing-service

GROUP            TOPIC          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
billing-service  order-events   0          128934          129500          566
billing-service  order-events   1          130220          130221          1
billing-service  order-events   2          125001          131890          6889

La partición 2 aquí es la que merece una alerta — su lag es órdenes de magnitud mayor que el de sus hermanas, lo que apunta a un hilo consumidor atascado, un mensaje poison-pill o una clave caliente, más que a un problema general de capacidad. En producción esto normalmente se conecta a un exporter de Prometheus o a una herramienta como Burrow, y se alerta sobre la tasa de cambio del lag, no solo su valor absoluto — un lag estable de 10.000 que no crece es una cuestión de capacidad que puedes planificar; un lag que sube 1.000/minuto es un incidente.

#Aplicar backpressure

Backpressure significa desacelerar (o pausar explícitamente) la ingesta cuando tu capacidad de procesar cae por debajo de tu capacidad de recibir — la alternativa es un buffer interno sin límite que eventualmente agota la memoria del consumidor. Algunas técnicas concretas, que normalmente se combinan:

• Prefetch acotado. Configura max.poll.records (Kafka) o el prefetch count de un canal (RabbitMQ) para que un solo poll nunca traiga más de lo que puedes sostener y procesar de forma segura — en vez de aceptar un lote sin límite y construir tú mismo un backlog en memoria.
• Pause/resume explícito. Cuando una cola interna delante de un pool de workers se llena, llama a consumer.pause() sobre las particiones afectadas para que el cliente deje de traer más registros de ellas (mientras sigue enviando heartbeats para mantenerse en el grupo), y consumer.resume() una vez que el backlog baja de un umbral.
• Colas internas acotadas. Dimensiona explícitamente la cola entre el loop de poll y el pool de workers, y haz que ofrecerle un elemento bloquee o falle rápido en vez de crecer sin límite.
• Escalar consumidores hasta el número de particiones. Agregar un consumidor más allá del número de particiones no hace nada (ver la pregunta sobre la arquitectura de Kafka) — si necesitas más throughput del que permite el número de particiones, tienes que aumentar las particiones, aceptando el trade-off de orden que eso implica.
• Agrupar (batch) las llamadas aguas abajo. Agrupar escrituras a base de datos o llamadas HTTP dentro del handler (en vez de un round trip por mensaje) suele elevar el throughput efectivo lo suficiente como para eliminar el lag sin añadir infraestructura.
• Circuit-break la dependencia aguas abajo. Si el lag lo causa un sistema externo lento o fallando, seguir golpeándolo empeora las cosas; un circuit breaker que falla rápido y reintenta después protege tanto al consumidor como al servicio aguas abajo.

Ajustar líneas largas
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "billing-service");
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);   // prefetch acotado por poll
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("order-events"));

BlockingQueue<ConsumerRecord<String, String>> inFlight = new ArrayBlockingQueue<>(500);
Set<TopicPartition> paused = ConcurrentHashMap.newKeySet();

while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));

    for (ConsumerRecord<String, String> record : records) {
        // Backpressure: si el pool de workers no da abasto, pausamos la partición
        // en vez de acumular registros en memoria sin límite.
        boolean accepted = inFlight.offer(record, 200, TimeUnit.MILLISECONDS);
        if (!accepted) {
            TopicPartition tp = new TopicPartition(record.topic(), record.partition());
            if (paused.add(tp)) {
                consumer.pause(List.of(tp));
                log.warn("Pausando {} - cola interna llena, el consumidor se está atrasando", tp);
            }
        }
    }

    if (!paused.isEmpty() && inFlight.remainingCapacity() > 250) {
        consumer.resume(paused);
        log.info("Reanudando {} particiones - backlog drenado", paused.size());
        paused.clear();
    }

    consumer.commitSync(currentOffsets(records)); // solo confirmar lo que los workers ya procesaron
}

Nota que pause()/resume() operan sobre particiones que el consumidor ya posee — detienen al cliente de traer nuevos registros de esa partición sin abandonar el consumer group, lo cual mantiene los rebalances (y la disrupción que causan en todas las demás particiones) completamente fuera de la ecuación.

🧠 Check your understanding
0/3 · 0/3 answered
1. Un consumer group se ve saludable en general (lag promedio bajo), pero una de ocho particiones tiene un lag de 500.000 y sigue subiendo mientras las demás están cerca de cero. ¿Cuál es la causa más probable?
Todo el clúster está sub-aprovisionadoEl topic necesita más réplicasUn mensaje poison-pill o una clave caliente está atascada en esa partición, congelando su consumidor mientras las demás procesan con normalidadKafka está perdiendo mensajes de esa partición
2. ¿Por qué es preferible llamar a consumer.pause() sobre una partición en vez de bloquear dentro del loop de poll mientras se espera una llamada lenta aguas abajo?
pause() borra el backlog para que nunca haya que procesarloNo hay diferencia entre ambos enfoquesBloquear el loop arriesga exceder max.poll.interval.ms, lo que expulsa al consumidor del grupo y dispara un rebalance que afecta a todas las particionespause() aumenta automáticamente el número de particiones
3. Necesitas más throughput de consumo en un topic que ya tiene un consumidor por partición. ¿Cuál es el siguiente paso correcto?
Agregar más instancias de consumidor al mismo grupoAumentar max.poll.records para que cada poll devuelva más datosBajar session.timeout.msAumentar el número de particiones (aceptando el trade-off de orden), ya que estás en el techo de paralelismo dado por el número de particiones