Fundamentos de la latencia en el producer

En arquitecturas de datos en tiempo real, el productor de Kafka constituye el punto de entrada. Su eficiencia determina la velocidad con la que los eventos llegan a los brokers. La latencia de envío abarca desde que la aplicación invoca send() hasta que recibe la confirmación del broker. Factores como la serialización, el ancho de banda de red y el tiempo de procesamiento del broker influyen en este intervalo.

El búfer de registros y buffer.memory

Antes de viajar por la red, cada registro se almacena en una cola interna de memoria. El parámetro buffer.memory fija el tamaño total de esa cola. Cuando la aplicación produce más rápido de lo que la red puede evacuar los datos, el búfer amortigua la diferencia.

Recomendaciones de tamaño

• Calcular el volumen de mensajes que podría acumularse durante el tiempo de recuperación ante un pico.
• Mantener el valor por encima de batch.size.
• Evitar reservas excesivas que presionen el recolector de basura de la JVM.

Cálculo orientativo

Si se generan 800 mensajes por segundo, con un tamaño medio de 1,5 KB, y se desea absorber 4 s de desfase, el búfer mínimo sería:

800 × 4 × 1,5 KB = 4,8 MB

Añadiendo margen para cabeceras de protocolo, un valor práctico sería de 6–8 MB.

Ejemplo de configuración

Properties cfg = new Properties();
cfg.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092");
cfg.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
cfg.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
cfg.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 8 * 1024 * 1024); // 8 MB
cfg.put(ProducerConfig.BATCH_SIZE_CONFIG, 32 * 1024);           // 32 KB
cfg.put(ProducerConfig.ACKS_CONFIG, "1");

KafkaProducer<String, String> producer = new KafkaProducer<>(cfg);

Agregación controlada con linger.ms

linger.ms define el tiempo máximo que el productor esperará para llenar un lote antes de enviarlo. Su valor por defecto es 0, lo que implica que cada mensaje se envía tan pronto como sea posible.

Compromiso entre latencia y throughput

• Valores bajos (0–2 ms): latencia reducida, pero más peticiones de red y mayor sobrecarga de protocolo.
• Valores medios (5–20 ms): equilibrio entre latencia aceptable y mejor aprovechamiento del ancho de banda.
• Valores altos (50–100 ms): máximo agrupamiento, ideal para ingestas masivas donde la latencia no es crítica.

Interacción entre buffer.memory, batch.size y linger.ms

El envío de un lote se desencadena por dos condiciones: el lote alcanza batch.size o transcurre linger.ms. El búfer total debe ser suficiente para retener los lotes pendientes mientras se forman y se transmiten.

Combinaciones típicas

Clase de prueba: impacto del tamaño del búfer

El siguiente programa compara dos configuraciones de búfer midiendo el tiempo total de inyección de 10 000 registros.

package com.ejemplo.kafka.productor;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.*;

public class PruebaMemoriaProductor {

    private static final String TOPICO = "metricas-buffer";
    private static final int REGISTROS = 10_000;
    private static final int HILOS = 2;

    public static void main(String[] args) throws InterruptedException {
        ejecutarEscenario(2 * 1024 * 1024, "buffer_2MB");
        ejecutarEscenario(16 * 1024 * 1024, "buffer_16MB");
    }

    private static void ejecutarEscenario(int memoriaBuffer, String etiqueta) throws InterruptedException {
        System.out.printf("%nIniciando escenario: %s%n", etiqueta);

        Properties cfg = new Properties();
        cfg.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        cfg.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        cfg.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        cfg.put(ProducerConfig.BUFFER_MEMORY_CONFIG, memoriaBuffer);
        cfg.put(ProducerConfig.BATCH_SIZE_CONFIG, 1024);
        cfg.put(ProducerConfig.LINGER_MS_CONFIG, 5);
        cfg.put(ProducerConfig.ACKS_CONFIG, "1");

        KafkaProducer<String, String> producer = new KafkaProducer<>(cfg);
        ExecutorService pool = Executors.newFixedThreadPool(HILOS);
        CountDownLatch barrera = new CountDownLatch(HILOS);

        long inicio = System.currentTimeMillis();

        for (int h = 0; h < HILOS; h++) {
            final int idHilo = h;
            pool.submit(() -> {
                try {
                    for (int r = 0; r < REGISTROS / HILOS; r++) {
                        String clave = "sensor_" + idHilo;
                        String valor = "lectura_" + r + "_" + System.nanoTime();
                        ProducerRecord<String, String> rec =
                                new ProducerRecord<>(TOPICO, clave, valor);
                        producer.send(rec, (meta, err) -> {
                            if (err != null) {
                                System.err.println("Error: " + err.getMessage());
                            }
                        });
                    }
                } finally {
                    barrera.countDown();
                }
            });
        }

        barrera.await(60, TimeUnit.SECONDS);
        long duracion = System.currentTimeMillis() - inicio;
        System.out.printf("Escenario %s finalizado en %d ms%n", etiqueta, duracion);

        producer.close();
        pool.shutdownNow();
    }
}

Clase de prueba: exploración de linger.ms

El sgiuiente ejemplo recorre varios valores de linger.ms y calcula métricas agregadas de latencia y throughput.

package com.ejemplo.kafka.productor;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

public class EvaluacionLingerMs {

    private static final String TOPICO = "metricas-linger";
    private static final int REGISTROS = 5_000;
    private static final int PARTICIPANTES = 2;

    private static final AtomicLong totalNs = new AtomicLong(0);
    private static final AtomicLong exitos = new AtomicLong(0);
    private static final AtomicLong fallos = new AtomicLong(0);

    public static void main(String[] args) throws InterruptedException {
        int[] valoresLinger = {0, 2, 10, 50};
        for (int linger : valoresLinger) {
            medirConfiguracion(linger);
        }
    }

    private static void medirConfiguracion(int lingerMs) throws InterruptedException {
        totalNs.set(0);
        exitos.set(0);
        fallos.set(0);

        Properties cfg = new Properties();
        cfg.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        cfg.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        cfg.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        cfg.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 8 * 1024 * 1024);
        cfg.put(ProducerConfig.BATCH_SIZE_CONFIG, 16 * 1024);
        cfg.put(ProducerConfig.LINGER_MS_CONFIG, lingerMs);
        cfg.put(ProducerConfig.ACKS_CONFIG, "1");

        KafkaProducer<String, String> producer = new KafkaProducer<>(cfg);
        ExecutorService pool = Executors.newFixedThreadPool(PARTICIPANTES);
        CountDownLatch barrera = new CountDownLatch(PARTICIPANTES);

        long t0 = System.currentTimeMillis();

        for (int h = 0; h < PARTICIPANTES; h++) {
            final int id = h;
            pool.submit(() -> {
                try {
                    for (int i = 0; i < REGISTROS / PARTICIPANTES; i++) {
                        String clave = "dispositivo_" + id;
                        String valor = "evento_" + i;
                        ProducerRecord<String, String> rec =
                                new ProducerRecord<>(TOPICO, clave, valor);

                        long inicio = System.nanoTime();
                        producer.send(rec, (metadata, ex) -> {
                            long fin = System.nanoTime();
                            totalNs.addAndGet(fin - inicio);
                            if (ex != null) {
                                fallos.incrementAndGet();
                            } else {
                                exitos.incrementAndGet();
                            }
                        });
                    }
                } finally {
                    barrera.countDown();
                }
            });
        }

        barrera.await(60, TimeUnit.SECONDS);
        long tiempoTotalMs = System.currentTimeMillis() - t0;

        long enviados = exitos.get();
        double latenciaMediaMs = (enviados > 0)
                ? (double) totalNs.get() / enviados / 1_000_000
                : 0;
        double throughput = (tiempoTotalMs > 0)
                ? (double) enviados / tiempoTotalMs * 1000
                : 0;

        System.out.printf(
            "%nlinger.ms=%d -> latencia media: %.2f ms, throughput: %.2f msg/s, fallos: %d%n",
            lingerMs, latenciaMediaMs, throughput, fallos.get());

        producer.close();
        pool.shutdownNow();
    }
}

Otros parámetros que afectan la latencia

• acks: con 0 se elimina la espera de confirmación, pero se pierde durabilidad; all maximiza la fiabilidad a costa de latencia.
• compression.type: snappy o lz4 reducen el tráfico de red a expensas de CPU.
• max.in.flight.requests.per.connection: valores altos aumentan el paralelismo, pero también el riesgo de reordenamiento si se reintentan envíos.
• retries y retry.backoff.ms: influyen en la latencia percibida ante fallos transitorios.

Diagnóstico de problemas comunes

Búfer saturado

Síntoma: excepciones de tipo BufferExhaustedException o bloqueos en send(). Causas habituales: buffer.memory insuficiente, brokers lentos o red congestionada. Acciones: ampliar el búfer, distribuir la carga entre varios productores, o revisar la capacidad del clúster.

Latencia inesperadamente alta

Causas frecuentes: linger.ms demasiado elevado, acks=all sin necesidad, compreisón intensiva en CPU, o latencia de red elevada. Acciones: reducir linger.ms, cambiar a acks=1 si el nivel de durabilidad lo permite, y elegir algoritmos de compresión ligeros.

Throughput bajo

Causas frecuentes: batch.size o linger.ms muy pequeños. Acciones: aumentar ambos parámetros y habilitar compresión.

Estrategia de validación

Antes de poner en producción una configuración, conviene ejecutar pruebas con kafka-producer-perf-test.sh o con programas Java propios, recopilando al menos:

• Latencia media y percentiles p95/p99.
• Mensajes por segundo.
• Tasa de errores.
• Uso de CPU y memoria del proceso productor.
• Latencia de red hacia los brokers.

Partir de un valor conservador y variar únicamente una variable a la vez facilita identificar el punto óptimo para la carga real.