MicroBatching en consumidores de Kafka

Algunas veces montamos sistemas más complejo de lo necesario porque desconocemos cómo funciona la tecnología que usamos.

Por ejemplo en la oficina un proyecto estaba proponiendo en un escenario con una etl en tiempo real que mueve datos de una base de datos a otro sistema usando kafka streams y kafka connect.

No saber como funciona la el consumidor de kafka, puede hacer que si el comportamiento por defecto no encaja, se aumente innecesariamente la complejidad del sistema que se implementa.

En este caso, el problema viene cuando en la base de datos de origen se realizan batches por la noche y en ese caso necesitan hacer una llamada por batch en vez de uno en uno para escribir los datos en el sistema origen.

El api que tiene que usar el proyecto para enviar los datos al otro sistema no tiene un buen rendimiento y han habilitado un método que permite el envío por lotes. Así que en vez de hacer una llamada por cada mensaje los puedes agrupar y mandar un lote.

Así que el equipo realiza una aplicación de kafka streams que dependiendo a qué hora se procesa el mensaje la salida se escribe en un mensaje u otro. Como podemos ver en el diagrama, cuando los mensajes se procesan por la noche, la información la mandan a un bulk topic que serán procesados por un conector de kafka connect manda por lotes los mensajes. Mientras que por el día el flujo de datos con menos intensidad continua por otro topic y otro sink

📝 Esto no es una buena idea

En esta solución hay varias cosas que pintan mal. Primero dos topics iguales, dos conectores… así que tenemos un sistema más complejo de lo esperado, con dos conectores en vez de uno que mantener y además dos topics con la misma estructura y posiblemente los mismos datos que generan un montón de dudas, como imposible asegurar un orden o coordinar la carga, imaginaos que el conector del bulk topic se atasca y deja de procesar (por el motivo qué sea) pasa ¿Qué pasa cuando acaba la noche y el conector de normal funciona?

Primero, no creo que sea necesario dos topics, si todos los mensajes representan lo mismo y tienen el mismo formato de datos… ¿para qué dos topics? Por otro lado, en vez de tener dos conectores, se puede tener uno y en función de cuantos mensajes haya leído el conector se puede mandar un batch o de uno en uno. Pero… ¿cómo controlo cuantos mensajes puede leer el consumidor de kafka? En la documentación podemos leer como se comporta el consumidor. Dónde nos explican que el consumidor sigue una estrategia de polling

The consumer provides two configuration settings to control the behavior of the poll loop:

• max.poll.interval.ms: By increasing the interval between expected polls, you can give the consumer more time to handle a batch of records returned from poll(long). The drawback is that increasing this value may delay a group rebalance since the consumer will only join the rebalance inside the call to poll. You can use this setting to bound the time to finish a rebalance, but you risk slower progress if the consumer cannot actually call poll often enough.
• max.poll.records: Use this setting to limit the total records returned from a single call to poll. This can make it easier to predict the maximum that must be handled within each poll interval. By tuning this value, you may be able to reduce the poll interval, which will reduce the impact of group rebalancing.

Así que si sabemos cuantos mensajes como máximo podemos enviar al api cuando procesamos por lotes podemos fijarlo en max.poll.records y según la calidad del servicio y la latencia que queramos tener podemos decir cuanto tiempo espera para leer los mensajes de kafka el consumidor con max.poll.interval.ms

Solo hay un truco, para modificar la configuración de los conectores en un cluster de kafka connect hay que habilitar la posibilidad de que el conector sobrescriba las configuraciones del cliente de kafka y poner en la configuración del conector los las propiedades anteriormente citadas con el prefijo consumer.override. Y aun se podría optimizar más y podemos hacer un sink más inteligente, pero con estas sencillas configuraciones se puede simplificar el sistema quedando algo como esto:

La configuración de los consumidores de kafka es algo que puede parecer fácil al principio pero puede llegar a ser muy complejo, un deep dive del consumidor (o de un sink connector de Kafka Connect) queda fuera del alcance de este post.

El objetivo de este post es explicar que siempre hay que profundizar un poco en la tecnología que se está usando y por otro lado siempre es buena idea leer la documentación oficial de los proyectos, posiblemente encuentres la explicación que necesitas