Cómo hacer compatible Kafka Consumer con Gevent en Python

La gestión de tareas asíncronas mediante Gevent mejora la escalabilidad y la eficiencia de recursos para sistemas distribuidos. Sin embargo, utilizar esta herramienta con Kafka puede suponer un reto. 

En DoorDash, muchos servicios están basados en Python, incluidas las tecnologías RabbitMQ y Celery, que fueron fundamentales para el sistema de cola de tareas asíncronas de nuestra plataforma. También aprovechamos Gevent, una biblioteca de concurrencia basada en coroutine, para mejorar aún más la eficiencia de nuestras operaciones de procesamiento de tareas asíncronas. A medida que DoorDash sigue creciendo, nos hemos enfrentado a retos de escalabilidad y hemos encontrado incidentes que nos impulsaron a sustituir RabbitMQ y Celery por Apache Kafka, una plataforma de streaming de eventos distribuidos de código abierto que ofrece una fiabilidad y escalabilidad superiores. 

Sin embargo, al migrar a Kafka, descubrimos que Gevent, la herramienta que utilizamos para el procesamiento de tareas asíncronas en nuestro sistema de punto de venta (TPV), no es compatible con Kafka. Esta incompatibilidad se produce porque utilizamos Gevent para parchear nuestras bibliotecas de código Python para realizar E/S asíncronas, mientras que Kafka se basa en librdkafka, una biblioteca C. El consumidor de Kafka bloquea la E/S de la librería C y no puede ser parcheado por Gevent de la forma asíncrona que buscamos.

Resolvimos este problema de incompatibilidad asignando manualmente Gevent a un hilo greenlet, ejecutando el procesamiento de tareas del consumidor Kafka dentro del hilo Gevent, y sustituyendo la E/S bloqueante de la tarea del consumidor por la versión de Gevent de la llamada "bloqueante" para lograr la asincronía. Las pruebas de rendimiento y los resultados reales de producción han demostrado que nuestro consumidor Kafka funciona sin problemas con Gevent, y supera al trabajador de tareas Celery/Gevent que teníamos antes, especialmente cuando se trata de tiempo de E/S pesado, que hacía que los servicios aguas abajo fueran lentos. 

¿Por qué pasar de RabbitMQ/Celery a Kafka con Gevent?

Para evitar una serie de interrupciones derivadas de nuestra lógica de procesamiento de tareas, varios equipos de ingeniería de DoorDash migraron de RabbitMQ y Celery a una solución Kafka personalizada. Aunque los detalles se pueden encontrar en este artículo, aquí hay un breve resumen de las ventajas de pasar a Kafka:  

• Kafka es una plataforma distribuida de streaming de eventos altamente fiable y disponible. También es famosa por su escalabilidad horizontal y el manejo de datos de producción a escala masiva.
• Kafka tiene una gran tolerancia a fallos porque se evita la pérdida de datos con la replicación de particiones.
• Como Kafka es un sistema de mensajería distribuido/pub-sub, su implementación encaja en el movimiento más amplio hacia los microservicios que se ha puesto en marcha en DoorDash. 
• En comparación con Celery, Kafka tiene mejor capacidad de observación y eficacia operativa, y nos ha ayudado a resolver los problemas e incidencias que encontrábamos cuando utilizábamos Celery.

Desde la migración a Kafka, muchos equipos de DoorDash han observado mejoras de fiabilidad y escalabilidad. Para obtener beneficios similares, nuestro equipo de comerciantes se preparó para migrar su servicio POS a Kafka. Lo que complica esta migración es el hecho de que los servicios de nuestro equipo también utilizan Gevent, porque:

• Gevent es una librería Python basada en coroutinas y E/S no bloqueantes. Con Gevent podemos procesar tareas pesadas de E/S de red de forma asíncrona sin que se bloqueen en espera de E/S, mientras seguimos escribiendo código de forma síncrona. Para saber más sobre nuestra implementación original de Gevent, lee este artículo del blog.
• Gevent puede parchear fácilmente código de aplicación existente o una biblioteca de terceros para E/S asíncrona, facilitando así su uso para nuestros servicios basados en Python.
• Gevent tiene una ejecución ligera a través de greenlets, y funciona bien cuando se escala con la aplicación. 
• Nuestros servicios tienen pesadas operaciones de E/S de red con partes externas como comerciantes, cuyas API pueden tener una latencia larga y puntiaguda, que no controlamos. Por tanto, necesitamos un procesamiento de tareas asíncrono para mejorar la utilización de los recursos y la eficiencia del servicio.
• Antes de implantar Gevent, solíamos sufrir cuando un socio importante tenía una interrupción, lo que podía afectar al rendimiento de nuestro propio servicio. 

Dado que Gevent es un componente fundamental para ayudarnos a conseguir un alto rendimiento en el procesamiento de tareas, queríamos obtener las ventajas de migrar a Kafka y mantener los beneficios de utilizar Gevent.

Los nuevos retos de la migración a Kafka 

Cuando empezamos a migrar de Celery a Kafka, nos enfrentamos a nuevos retos al intentar mantener Gevent intacto. En primer lugar, queríamos mantener el alto rendimiento existente en el procesamiento de tareas que permitía Gevent, pero no pudimos encontrar una biblioteca Kafka Gevent lista para usar ni ningún recurso en línea para combinar Kafka y Gevent. 

Estudiamos cómo la aplicación monolítica de DoorDash migró de Celery a Kafka, y descubrimos que esos casos de uso utilizaban un proceso dedicado por cada tarea. En nuestros servicios y con nuestros casos de uso, dedicar un proceso por tarea provocaría un consumo excesivo de recursos en comparación con la utilización de los hilos de Gevent. Simplemente no podíamos replicar el trabajo de migración que se había hecho antes en DoorDash, y tuvimos que elaborar una implementación más novedosa para nuestros casos de uso, que implicaba operar con Gevent sin la pérdida de eficiencia.

Cuando investigamos nuestra propia implementación de consumidor Kafka con Gevent, identificamos un problema de incompatibilidad: como la librería confluent-kafka-python que usamos está basada en la librería C librdkafka, sus llamadas de bloqueo no pueden ser parcheadas por Gevent porque Gevent sólo funciona con código y librerías Python. Si ingenuamente reemplazamos el trabajador Celery con un consumidor Kafka para sondear los mensajes de la tarea, nuestros hilos Gevent de procesamiento de tareas existentes serán bloqueados por la llamada de sondeo del consumidor Kafka, y perderemos todos los beneficios de usar Gevent.

while True:
   message = consumer.poll(timeout=TIME_OUT)
   if not message:
       continue

Figura 2: Este fragmento de código es una implementación típica de consumidor Kafka con un tiempo de espera definido en el sondeo de mensajes. Sin embargo, bloquea los hilos de Gevent ya que librdkafka realiza el tiempo de espera.

Sustitución de la llamada de bloqueo de Kafka por una llamada asíncrona de Gevent

Estudiando artículos online sobre un problema similar al trabajar con Kafka producer y Gevent, se nos ocurrió una solución para resolver el problema de incompatibilidad entre el consumidor Kafka y Gevent: cuando el consumidor Kafka sondea mensajes, ponemos el tiempo de espera de bloqueo de Kafka a cero, lo que ya no bloquea nuestros hilos de Gevent. 

En el caso de que no haya ningún mensaje disponible para sondear, con el fin de ahorrar el ciclo de CPU en el bucle de sondeo de mensajes, añadimos un mensaje gevent.sleep(timeout) llamada. De esta forma, podemos cambiar de contexto para realizar el trabajo de otros hilos mientras el hilo consumidor de Kafka está en reposo. Como Gevent realiza la suspensión, otros hilos de Gevent no se bloquearán mientras esperamos el siguiente sondeo de mensajes del consumidor.

while True:
   message = consumer.poll(timeout=0)
   if not message:
       gevent.sleep(TIME_OUT)
       continue

Figura 3: Establecer el tiempo de espera de sondeo de mensajes del consumidor de Kafka a cero ya no bloquea los hilos de Gevent.

Una posible desventaja de hacer este cambio de contexto manual del hilo Gevent es que si interferimos con el ciclo de consumo de mensajes de Kafka, podemos sacrificar cualquier optimización que provenga de la biblioteca Kafka. Sin embargo, a través de pruebas de rendimiento, no hemos visto degradaciones después de hacer ese tipo de cambios, y en realidad podría ver mejoras de rendimiento utilizando Kafka en comparación con Celery.

Comparación del rendimiento: Kafka frente a Celery

El siguiente gráfico muestra la comparación de rendimiento, en tiempo de ejecución, entre Kafka y Celery. Celery y Kafka muestran resultados similares en cargas pequeñas, pero Celery es relativamente sensible a la cantidad de trabajos concurrentes que ejecuta, mientras que Kafka mantiene el tiempo de procesamiento casi igual independientemente de la carga. El número máximo de trabajos que se ejecutaron de forma concurrente en las pruebas es de 6.000 y Kafka muestra un gran rendimiento incluso con retrasos de E/S en los trabajos, mientras que el tiempo de ejecución de tareas de Celery aumenta notablemente hasta 140 segundos. Mientras que Celery es competitivo para pequeñas cantidades de trabajos sin tiempo de E/S, Kafka supera a Celery para grandes cantidades de trabajos concurrentes, especialmente cuando hay retrasos de E/S.

Resultados

Migrar de Celery a Kafka sin dejar de utilizar Gevent nos permite disponer de una solución de colas de tareas más fiable a la vez que mantenemos un alto rendimiento. Los experimentos de rendimiento anteriores muestran resultados prometedores para situaciones de alto volumen y alta latencia de E/S. Hasta ahora hemos estado ejecutando Kafka consumer con Gevent durante un par de meses en producción, y hemos visto un alto rendimiento fiable sin la recurrencia de los problemas que vimos antes cuando usábamos Celery. 

Conclusión

Usar Kafka con Gevent es una potente combinación. Kafka se ha probado a sí mismo y ha ganado popularidad como un bus de mensajería y una solución de colas, mientras que Gevent es una poderosa herramienta para mejorar el rendimiento de servicios Python de E/S pesada. Desafortunadamente, no pudimos encontrar ninguna biblioteca disponible para combinar Kafka y Gevent juntos, posiblemente debido a la razón de que Gevent no funciona con la biblioteca C librdkafka en la que se basa Kafka. En nuestro caso, pasamos por la lucha, pero nos alegramos de encontrar una solución que funcionaba para mezclar los dos. Para otras empresas, si el alto rendimiento, escalabilidad y fiabilidad son las propiedades deseadas para sus aplicaciones Python que requieren un bus de mensajería, Kafka con Gevent podría ser la respuesta. 

Agradecimientos

Los autores agradecen a Mansur Fattakhov, Hui Luan, Patrick Rogers, Simone Restelli y Adi Sethupat sus aportaciones y consejos durante este proyecto.