Como tornar o consumidor Kafka compatível com o Gevent em Python

A gestão assíncrona de tarefas utilizando o Gevent melhora a escalabilidade e a eficiência de recursos para sistemas distribuídos. No entanto, a utilização desta ferramenta com o Kafka pode ser um desafio. 

Na DoorDash, muitos serviços são baseados em Python, incluindo as tecnologias RabbitMQ e Celery, que foram fundamentais para o sistema de fila de tarefas assíncronas da nossa plataforma. Também aproveitamos o Gevent, uma biblioteca de concorrência baseada em corrotinas, para melhorar ainda mais a eficiência das nossas operações de processamento de tarefas assíncronas. À medida que a DoorDash continua a crescer, enfrentámos desafios de escalabilidade e deparámo-nos com incidentes que nos levaram a substituir o RabbitMQ e o Celery pelo Apache Kafka, uma plataforma de streaming de eventos distribuídos de código aberto que oferece fiabilidade e escalabilidade superiores. 

No entanto, ao migrar para o Kafka, descobrimos que o Gevent, a ferramenta que utilizamos para o processamento assíncrono de tarefas no nosso sistema de ponto de venda (POS), não é compatível com o Kafka. Essa incompatibilidade ocorre porque usamos o Gevent para corrigir nossas bibliotecas de código Python para executar E/S assíncronas, enquanto o Kafka é baseado na librdkafka, uma biblioteca C. O consumidor do Kafka bloqueia a E/S da biblioteca C e não pode ser corrigido pelo Gevent da maneira assíncrona que estamos procurando.

Resolvemos esse problema de incompatibilidade alocando manualmente o Gevent em um thread greenlet, executando o processamento da tarefa do consumidor Kafka dentro do thread Gevent e substituindo a E/S de bloqueio da tarefa do consumidor pela versão do Gevent da chamada "blocking" para obter assincronia. Os testes de desempenho e os resultados reais de produção mostraram que o nosso consumidor Kafka funciona sem problemas com o Gevent e supera o desempenho do trabalhador de tarefas Celery/Gevent que tínhamos antes, especialmente quando lidamos com tempo de E/S pesado, o que tornava os serviços a jusante lentos. 

Porquê mudar do RabbitMQ/Celery para o Kafka com Gevent?

Para evitar uma série de interrupções decorrentes da nossa lógica de processamento de tarefas, várias equipas de engenharia do DoorDash migraram do RabbitMQ e do Celery para uma solução Kafka personalizada. Embora os detalhes possam ser encontrados neste artigo, aqui está um breve resumo das vantagens de migrar para o Kafka:  

• O Kafka é uma plataforma de transmissão de eventos distribuída que é altamente fiável e disponível. É também famosa pela sua escalabilidade horizontal e pelo tratamento de dados de produção em grande escala.
• O Kafka tem uma grande tolerância a falhas porque a perda de dados é evitada com a replicação de partições.
• Como o Kafka é um sistema de mensagens distribuído/pub-sub, a sua implementação enquadra-se na mudança mais alargada para os microsserviços que foi implementada na DoorDash. 
• Em comparação com o Celery, o Kafka tem melhor observabilidade e eficiência operacional e ajudou-nos a resolver os problemas e incidentes que encontrámos quando utilizámos o Celery.

Desde a migração para o Kafka, muitas equipas da DoorDash registaram melhorias na fiabilidade e na escalabilidade. Para obter benefícios semelhantes, a nossa equipa de comerciantes preparou-se para migrar o seu serviço de POS para o Kafka. A complicar esta migração está o facto de os serviços da nossa equipa também utilizarem o Gevent, porque:

• Gevent é uma biblioteca Python baseada em corrotinas e E/S sem bloqueio. Com Gevent podemos processar tarefas pesadas de E/S de rede de forma assíncrona sem que elas fiquem bloqueadas à espera de E/S, enquanto continuamos a escrever código de forma síncrona. Para saber mais sobre a nossa implementação original do Gevent, leia este artigo do blogue.
• O Gevent pode facilmente fazer o " monkey-patch" do código da aplicação existente ou de uma biblioteca de terceiros para E/S assíncronas, tornando-o assim fácil de utilizar para os nossos serviços baseados em Python.
• O Gevent tem uma execução leve através de greenlets e tem um bom desempenho quando é escalado com a aplicação. 
• Os nossos serviços têm operações de E/S de rede pesadas com entidades externas, como comerciantes, cujas API podem ter uma latência longa e irregular, que não controlamos. Assim, precisamos de processamento assíncrono de tarefas para melhorar a utilização dos recursos e a eficiência do serviço.
• Antes de implementar a Gevent, costumávamos sofrer quando um parceiro importante estava a ter uma interrupção, o que podia afetar o desempenho do nosso próprio serviço. 

Como o Gevent é um componente crítico para nos ajudar a alcançar um elevado rendimento de processamento de tarefas, queríamos obter as vantagens da migração para o Kafka e manter os benefícios da utilização do Gevent.

Os novos desafios da migração para o Kafka 

Quando começamos a migrar do Celery para o Kafka, enfrentamos novos desafios ao tentar manter o Gevent intacto. Em primeiro lugar, queríamos manter o alto rendimento do processamento de tarefas existente que era permitido pelo Gevent, mas não conseguimos encontrar uma biblioteca Kafka Gevent pronta para uso ou quaisquer recursos on-line para combinar o Kafka e o Gevent. 

Estudámos a forma como a aplicação monolítica do DoorDash migrou do Celery para o Kafka e descobrimos que esses casos de utilização estavam a utilizar um processo dedicado por cada tarefa. Em nossos serviços e com nossos casos de uso, dedicar um processo por tarefa causaria um consumo excessivo de recursos em comparação com a utilização dos threads do Gevent. Simplesmente não pudemos replicar o trabalho de migração que havia sido feito antes na DoorDash e tivemos que elaborar uma implementação mais recente para nossos casos de uso, que envolvia operar com o Gevent sem a perda de eficiência.

Quando analisámos a nossa própria implementação do consumidor Kafka com o Gevent, identificámos um problema de incompatibilidade: como a biblioteca confluent-kafka-python que usamos é baseada na biblioteca C librdkafka, as suas chamadas de bloqueio não podem ser corrigidas pelo Gevent porque o Gevent só funciona em código e bibliotecas Python. Se substituirmos ingenuamente o trabalhador do Celery por um consumidor Kafka para sondar as mensagens da tarefa, nossas threads Gevent de processamento de tarefas existentes serão bloqueadas pela chamada de sondagem do consumidor Kafka e perderemos todos os benefícios do uso do Gevent.

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

Figura 2: Este trecho de código é uma implementação típica de um consumidor Kafka com um tempo limite definido para a pesquisa de mensagens. No entanto, bloqueia as threads Gevent, uma vez que o timeout é efectuado pelo librdkafka.

Substituir a chamada de bloqueio do Kafka por uma chamada assíncrona do Gevent

Ao estudar artigos online sobre um problema semelhante quando se trabalha com o produtor Kafka e o Gevent, encontrámos uma solução para resolver o problema de incompatibilidade entre o consumidor Kafka e o Gevent: quando o consumidor Kafka sonda as mensagens, definimos o tempo limite de bloqueio do Kafka para zero, o que já não bloqueia as nossas threads Gevent. 

No caso de não haver nenhuma mensagem disponível para sondar, para economizar o ciclo da CPU no loop de sondagem de mensagens, adicionamos um gevent.sleep(timeout) chamada. Dessa forma, podemos alternar o contexto para executar o trabalho de outros threads enquanto o thread do consumidor Kafka está em sleep. Como o sleep é executado pelo Gevent, outros threads do Gevent não serão bloqueados enquanto aguardamos a próxima sondagem de mensagem do consumidor.

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

Figura 3: Definir o tempo limite de sondagem de mensagens do consumidor Kafka para zero já não bloqueia os threads Gevent.

Uma possível desvantagem de fazer essa troca manual de contexto de thread do Gevent é que, se interferirmos no ciclo de consumo de mensagens do Kafka, podemos sacrificar qualquer otimização que venha da biblioteca do Kafka. No entanto, por meio de testes de desempenho, não vimos degradações depois de fazer esses tipos de alterações e, na verdade, podemos ver melhorias de desempenho usando o Kafka em comparação com o Celery.

Comparação da taxa de transferência: Kafka vs Celery

O gráfico abaixo mostra a comparação da taxa de transferência, em tempo de execução, entre o Kafka e o Celery. O Celery e o Kafka mostram resultados semelhantes em cargas pequenas, mas o Celery é relativamente sensível à quantidade de trabalhos simultâneos que executa, enquanto o Kafka mantém o tempo de processamento quase o mesmo, independentemente da carga. O número máximo de tarefas que foram executadas simultaneamente nos testes é de 6.000 e o Kafka mostra um grande rendimento mesmo com atrasos de E/S nas tarefas, enquanto o tempo de execução da tarefa do Celery aumenta visivelmente até 140 segundos. Enquanto o Celery é competitivo para pequenas quantidades de trabalhos sem tempo de E/S, o Kafka supera o Celery para grandes quantidades de trabalhos simultâneos, especialmente quando há atrasos de E/S.

Resultados

A migração do Celery para o Kafka, embora continuando a utilizar o Gevent, permite-nos ter uma solução de enfileiramento de tarefas mais fiável, mantendo uma elevada taxa de transferência. As experiências de desempenho acima mostram resultados promissores para situações de alto volume e alta latência de E/S. Até agora, estamos a executar o consumidor Kafka com Gevent há alguns meses em produção e temos visto um rendimento fiável elevado sem a recorrência de problemas que vimos anteriormente quando usámos o Celery. 

Conclusão

Usar o Kafka com o Gevent é uma combinação poderosa. O Kafka se provou e ganhou popularidade como um barramento de mensagens e solução de enfileiramento, enquanto o Gevent é uma ferramenta poderosa para melhorar a taxa de transferência de serviços Python pesados de E/S. Infelizmente, não encontramos nenhuma biblioteca disponível para combinar Kafka e Gevent, possivelmente devido ao fato de que Gevent não funciona com a biblioteca C librdkafka na qual Kafka é baseado. No nosso caso, passámos por dificuldades, mas ficámos satisfeitos por encontrar uma solução funcional para combinar os dois. Para outras empresas, se a alta taxa de transferência, a escalabilidade e a confiabilidade forem as propriedades desejadas para seus aplicativos Python que exigem um barramento de mensagens, o Kafka com Gevent pode ser a resposta. 

Agradecimentos

Os autores gostariam de agradecer a Mansur Fattakhov, Hui Luan, Patrick Rogers, Simone Restelli e Adi Sethupat pelas suas contribuições e conselhos durante este projeto.