Uma introdução ao Apache Kafka, lições aprendidas em um ambiente de varejo

Este artigo, é um relato de experiências e ajustes na utilização de clusters Apache Kafka em um dos maiores varejistas do Brasil e de quais técnicas e processos que utilizamos ao lidar com clusters de Apache Kafka processando uma volumetria que passa de 5 milhões de mensagens por dia nos cenários de stream (eventos, processamento e consolidação) que são transacionais e 100 milhões de mensagens que são replicadas diariamente através de CDC do DB2 para o Apache Kafka para a disponibilização de dados para a baixa plataforma.

Segundo o Apache Kafka Report 2018, apresentado pela Confluent. O Apache Kafka é uma solução largamente utilizada tanto em empresas de pequeno porte como em empresas de muito grande porte.

• Como podemos ter 100% de certeza que os nossos dados estarão no cluster Kafka?
• Há a possibilidade de perder mensagens no Apache Kafka?
• Como podemos verificar se as mensagens persistidas no Apache Kafka realmente estão disponíveis para consumo?

Essas são algumas das perguntas que buscaremos responder neste artigo.

Introdução

Durante as rotinas do dia a dia somado às várias particularidades que são enfrentadas por times de desenvolvimento em nosso ambiente, no qual precisamos integrar, extrair e processar dados de sistemas Mainframes IBM, .Net e Java, o passar dos anos e as características empíricas do dia a dia nos fizeram entender e aprender que é necessário não apenas saber escrever e consumir um tópico no Apache Kafka mas também entender as particularidades inerentes a uma solução não tão simples, mas que é capaz de fazer um grande diferencial no dia a dia de uma empresa que processa em uma única Black Friday mais de 120 milhões de mensagens transacionais. O uso do Apache Kafka na Via Varejo remonta ao ano de 2014 e neste caminho, muitas lições foram aprendidas e são algumas delas que buscamos compartilhar neste artigo.

O nosso primeiro caso de uso com o Apache Kafka foi implementado para resolver problemas de integração entre a nossa plataforma de marketplace e o e-commerce o qual na época tínhamos os seguintes problemas:

• Lentidão na integração das ofertas dos lojistas marketplace com e-commerce que refletiam em:
  • Um preço ou disponibilidade de uma oferta demorava cerca de 8 horas para refletir no site.
  • O processo era síncrono, e muitas vezes o alto volume das integrações gerava indisponibilidade no site.
  • O time desligava as integrações, ligando somente à noite, para não haver risco de gerar indisponibilidade no site.
  • A maioria das atualizações eram feitas por meio de processos batch.

Esses problemas faziam com que os produtos dos lojistas marketplace que estivessem sem estoque fossem exibidos no site como disponível, ou pior, produtos com preços desatualizados.

O Apache Kafka nos permitiu tornar essas integrações assíncronas e em tempo real, ao invés de integrar com processos batch, passamos a gerar eventos cada vez que uma oferta era criada, atualizada ou deletada, possibilitando atualizações distribuídas, entregando mais throughput.

Atualmente utilizamos o Apache Kafka em vários cenários, em dias normais processamos centenas de milhares de pedidos. Para cada pedido criado, temos mais de 20 eventos relacionados, como: movimentações de estoque, trackings, processamentos de pagamento, notificações ao cliente, entre outros.

O relatório da Confluent de 2018, deixa claro que o aprendizado e o lidar no dia a dia com uma ferramenta como essa não é tão simples conforme as respostas coletadas e representadas na imagem a seguir e consequentemente a contratação de talentos que possam aproveitar todas as funcionalidades de uma solução como esta torna-se um pouco mais difícil.

Figura 1 - Taxa de dificuldade na contratação de profissionais especializados - Confluent Report 2018

A partir deste momento, vamos explicitar alguns dos detalhes técnicos e lições aprendidas nos últimos anos iniciando com um overview básico até chegar em aspectos mais avançados.

Sobre a arquitetura do Apache Kafka

A arquitetura do Apache Kafka é organizada em torno de alguns termos chaves, são eles: messages, topics, producers, consumers groups e os brokers.

O Apache Kafka é uma plataforma de stream (fluxo de dados) de alto throughput que desacopla os produtores de dados dos consumidores de dados. Nele, as mensagens são organizadas em tópicos, os tópicos são divididos em partições, e as partições são replicadas entre os nós denominados brokers.

Figura 2 - Exemplo de consumo distribuído

Messages (Mensagens)

No Apache Kafka as mensagens representam a unidade de dados, o registro do seu tópico. Toda mensagem é do tipo chave / valor, e elas são adicionadas sequencialmente ao final de um arquivo de log de uma partição e numeradas por offsets exclusivos. São também persistidas em disco e replicadas no cluster para evitar a perda de dados. As mensagens também possuem um cache de páginas em memória para melhorar a performance de leituras dos dados e ficam disponíveis nas partições até serem excluídas quando ocorrer um TTL ou após uma rotina de compactação.

Offsets são as posições das mensagens em uma partição de um tópico no Apache Kafka.

Topics (Tópicos)

Um tópico é constituído de no mínimo uma partição e zero ou mais réplicas seguidoras, as partições permitem que vários consumidores leiam um tópico em paralelo, permitindo assim uma alta taxa de processamento (figura 2, consumidor B). Essa é uma das características que diferencia o Apache Kafka dos outros MOM (Message-oriented Middlewares), colocando-o como uma solução distribuída de mensageria.

Figura 3 - Exemplo de um tópico com três partições - Fonte: Hortonworks Docs

Cada mensagem dentro de uma partição tem um identificador chamado offset, o offset mantém as mensagens ordenadas em uma sequência imutável, permitindo aos consumidores lerem a partir de qualquer ponto do offset.

Figura 4 - Consumidor A e B lendo a partir de diferentes offsets

Cada mensagem em um cluster Apache Kafka pode ser identificada exclusivamente por uma tupla que consiste no nome do tópico, na partição, e no offset da partição.

O Apache Kafka mantém as mensagens por um período de tempo configurável, o padrão é de até 7 dias. Se um tópico estiver configurado para manter as mensagens por no máximo 24 horas, e o consumidor ficar inativo por mais de um dia, o consumidor perderá as mensagens; se o consumidor ficar indisponível por uma hora, ele começará a ler a partir do último offset conhecido pelo seu consumer group no Apache Kafka.

Producers (Produtores)

Um produtor é qualquer aplicação que gera mensagens para serem persistidas no Apache Kafka. Um produtor é capaz de publicar mensagens em um ou mais tópicos de um cluster no Apache Kafka.

Consumers (Consumidores)

Os consumidores são os assinantes (leitores) das mensagens de um determinado tópico, a diferença fundamental entre o Apache Kafka e um MOM tradicional é que os consumidores nunca receberão as mensagens automaticamente, eles tem que pedir as mensagens, quando estiverem prontos para processar as mesmas.

Brokers

Segundo a documentação do Apache Kafka, um broker é:

Em um servidor Apache Kafka, um broker Apache Kafka e um nó Apache Kafka se referem ao mesmo conceito e são sinônimos (veja o scaladoc do KafkaServer)

E o conceito segundo uma visão de negócios para um broker de acordo com a Wikipédia temos:

Um broker é um indivíduo que organiza transações entre um comprador e um vendedor com o objetivo de receber uma comissão quando a transação é executada com sucesso.

Uma das principais responsabilidades de um broker é reunir vendedores e compradores. Sendo assim, um broker atua como uma terceira pessoa facilitadora entre um comprador e um vendedor.

Dito isto, um broker Apache Kafka recebe mensagens dos produtores e as armazena em disco, com uma chave exclusiva de offset. Um broker Apache Kafka permite que os consumidores busquem mensagens por tópico, consumer group, partição e offset. Brokers fazem parte de um cluster compartilhando informações entre si direta ou indiretamente, sendo que um dos brokers atua como controlador (controller).

O papel do controlador é:

• Monitorar a saúde dos brokers
• Eleger novos líderes, em caso de falha em um dos brokers
• Comunicar os novos líderes para os brokers

Consumer Group

Uma poderosa abstração proposta pelo Apache Kafka relacionada aos consumidores é a definição de grupo de consumidores (Consumer Groups). O consumer group é a junção de dois conceitos de sistemas de mensageria: enfileiramento e publish-subscribe.

Quando temos um consumer group consumindo um tópico, o Apache Kafka se comporta como um sistema de mensageria tradicional baseado em enfileiramento. Os consumidores dentro de um mesmo consumer group dividem entre si o acesso às partições do tópico, de forma que cada partição seja consumida por somente uma única instância de consumidor do grupo.

Complementando esse conceito, quando temos múltiplos consumer groups para um mesmo tópico, o Apache Kafka faz o broadcast das mensagens para todos os consumer groups, se comportando como um sistema de mensageria baseado em publish-subscribe.

Um consumer group serve para indicar que todas as mensagens, de todas as partições de um determinado tópico serão consumidas. Uma partição só pode ser consumida por uma única thread ou processo. Se por algum motivo houver mais consumidores que partições, os consumidores adicionais ficarão ociosos porque não haverá partições para ler. Caso existam mais partições do que consumidores, os consumidores receberão mensagens de várias partições (figura 5, aplicação Y rodando com 2 instâncias). Em um cenário em que houver números iguais de consumidores e partições, haverá uma distribuição de um para um, o que é ideal para um maior throughput (figura 5, aplicação Z com quatro instâncias).

Figura 5 - Aplicação Y e Z consumindo um mesmo tópico com 4 partições.

Caso o processo Y2 falhe, o Apache Kafka fará uma redistribuição fazendo com que o processo Y1 passe a ler também as partições P1 e P2. Quando o processo Y2 retornar o Apache Kafka fará novamente uma redistribuição de forma a balancear a carga entre os dois processos.

Observe na figura 6, a distribuição que há a partir dos brokers (servers), cada broker é o líder de uma partição e a carga normalmente é distribuída de forma igualitária entre os brokers.

Figura 6 - Replicações e ISRs

O Apache Zookeeper

Os criadores do Apache Kafka decidiram aproveitar o Apache ZooKeeper como um serviço de coordenação de cluster. O ZooKeeper fornece recursos de gerenciamento de configuração, sincronização distribuída, participação em grupo e recebimento de eleição de líder. Seu modelo de dados representa um conjunto de nós (chamados "znodes") organizados em estrutura de dados hierárquica em forma de árvore. Cada nó pode ter uma mistura de características diferentes - ser persistente, efêmero ou seqüencial - e conter payload arbitrário armazenado em um formato de array de bytes binários.

O ZooKeeper provê um mecanismo de monitoramento único que permite que os clientes recebam notificações, uma vez que o znode representando o caminho específico, ou qualquer um de seus filhos seja criado, atualizado ou removido. O Apache Kafka implementa a descoberta de serviços usando dados e monitoramentos efêmeros. Todo broker se inscreve em notificações no znode "/brokers/ids" para receber um evento sempre que seus pares saírem ou se juntarem ao cluster.

Quando um novo tópico é criado com um fator de replicação 3 e 10 partições, o controller elegerá um novo líder por partição, buscando distribuir as líderes de maneira ideal entre os brokers.

Para cada partição o controller:

• Atualizará o Zookeeper com o ISR e a líder.
• Enviará um comando LeaderAndISRCommand para cada broker que é uma réplica da partição, informando aos brokers o ISR e a líder.

Quando um broker líder ficar indisponível, o Zookeeper enviará uma notificação para o controller, e o controller elegerá uma nova líder. Após a eleição, o controller enviará o comando LeaderAndISRCommand para cada broker que é uma réplica da partição, informando aos brokers a nova líder e os novos ISR. Sempre que a ISR muda, o líder atualizar o Zookeeper da mudança.

O Zookeeper sempre está atualizado com o estado do cluster Kafka, se um líder falhar, o Zookeeper coordenará as notificações para o controller, para a transição do novo líder.

Figura 7 - Arquitetura de Consenso do Kafka

Recentemente foi aprovada uma proposta de remoção do Apache Zookeeper que está em discussão em melhorias propostas, que tem o objetivo de reduzir os pontos de fallhas causados pela dependência com o Apache ZooKeeper, facilitando também a configuração e a publicação do Apache Kafka.

Evitando falhas

A partir de agora demonstraremos com exemplos, cenários passíveis de perda de dados e como alguns ajustes simples de configuração podem mitigar essas falhas.

Como o Apache Kafka cuida das falhas:

• A proteção do Apache Kafka contra falhas ocorre por meio de replicação de dados;
• A unidade de replicação utilizada pelo Apache Kafka é a partição;
• Em um cluster, um broker sempre é designado como líder de uma ou mais partições;
• Os brokers secundários sempre são atualizados a partir do líder;
• Cada réplica de dados está armazenada em um broker diferente;
• O broker líder é o responsável por executar as escritas e leituras na lista de in-sync réplicas, que são replicadas para os brokers seguidores.

Nota: Não devemos confundir particionamento com replicação, a partição é a forma como escalamos um tópico, podendo ter centenas de partições para um único tópico em um cluster com apenas 3 brokers, assim cada partição teria 3 réplicas de seus dados. Maior quantidade de partições representam um maior throughput, permitindo que mais computadores disponíveis consumam um tópico em paralelo.

O Apache Kafka escolhe uma réplica de partição em um broker líder usando o Zookeeper. O broker com a partição líder controla todas as escritas e leituras das mensagens da partição. A partição líder é a responsável por propagar as mensagens para as suas réplicas nos diferentes brokers. Uma réplica seguidora (follower) que está em sincronia é chamada de ISR (in-sync replica). Se uma partição líder falhar, o Apache Kafka escolherá uma nova ISR para ser líder.

Nota: Os consumidores não consomem mensagens dos seguidores (followers), os seguidores só existem para redundância e failover.

O comando kafka-topics.sh --zookeeper your-zookeeper-cluster --describe nos fornece uma boa visão do estado de sincronia das replicações. Observe a figura abaixo:

Figura 8 - Visão dos BROKERS líderes e de seus seguidores.

Algumas terminologias importantes para compreensão do texto abaixo

Acked: O acked é uma mensagem retornada pelo Apache Kafka confirmando o recebimento da mesma.

• Uma vez recebida, o produtor não irá tentar enviar a mensagem novamente;
• A resposta depende das configurações do produtor.

Commited: É um estado que indica que uma mensagem está disponível para ser consumida.

• Ocorre somente quando a mensagem está em todos os ISRs;
• Uma vez disponível, os consumidores podem ler;
• replica.lag.time.max.ms: É um parametro que indica quanto tempo uma réplica indisponível irá impedir os consumidores de lerem as mensagens.

Commited Offsets: É o parâmetro que indica o último offset com a leitura confirmada pelo consumidor. Por padrão o consumidor nunca irá ler essa mensagem novamente.

Quando todos os brokers de um cluster estão disponíveis, os produtores e consumidores podem escrever e ler a partir da partição líder do tópico sem problemas. Infelizmente os líderes e as réplicas podem falhar e precisamos saber lidar com cada uma dessas situações.

Cluster Saudável

Figura 9 - Todos os brokers de um cluster disponíveis.

Um nó do cluster com falha

As escritas deixarão de chegar e ele não receberá mais mensagens das outras réplicas, ficando cada vez mais fora de sincronia com o líder.

Figura 10 - Falha na réplica 3

Conforme apresentado na figura 10, por algum motivo a réplica 3 parou, a mensagem 100 está confirmada commited e pode ser lida pelos consumidores, a mensagem 101 não pode ser lida pois ela ainda não foi confirmada commited, ou seja, não está presente em todos os ISR. Após a remoção do réplica 3 do ISR todas as mensagens se tornarão acked e commited.

Figura 11 - Falha na réplica 2

O que acontece quando uma segunda réplica falhar? Neste caso, ela não estará mais disponível para receber as mensagens e também ficará fora de sincronia com a líder.

Neste momento, apenas o líder está em sincronia. Na terminologia do Apache Kafka ainda temos uma réplica em sincronia, embora essa réplica seja a líder da partição.

Figura 12 - Falha em todas as réplicas

Nesse momento, a réplica 1 não pode mais receber dados, mas ela está sincronizada com tudo que foi possível receber. A réplica 2 tem alguns dados faltando, e a réplica 3, a primeira a ficar indisponível está com uma maior quantidade de dados faltando.

Caso as réplicas 2 ou 3 voltem a operar antes da líder, teremos perda de dados.

Figura 13 - Perda de dados

Quando uma réplica volta a estar operacional antes da líder, a mesma estará fora de sincronia e todos os dados escritos no período de tempo em que esta réplica estava indisponível serão perdidos após ela tornar-se a nova líder da partição. Quando as réplicas adicionais retornarem, elas verificarão se possuem mensagens confirmadas - commited messages - que não existam no novo líder, se houver, a réplica (ex-líder) apagará as mensagens em sua posse e se tornará a seguidora da nova líder. Ao eleger uma nova líder assim que possível, mensagens poderão ser perdidas, mas o tempo de inatividade (downtime) será minimizado com qualquer novo broker podendo ser o líder das partições do tópico.

Se uma réplica líder falhar, o controlador (controller) do Apache Kafka irá detectar a perda da líder a partir do pool do in-sync replicas e irá eleger uma nova líder. Isso pode demorar alguns segundos e os clientes do Apache Kafka - produtores e consumidores - poderão receber o erro de LeaderNotAvailable. Nenhum dado será perdido se os clientes controlarem essa possibilidade e estiverem configurados corretamente.

Nota: O Apache Kafka não enfrenta problemas com Split Brain. Ele utiliza um mecanismo de epoch number, que nada mais é do que um contador que é incrementado toda vez que um novo controlador é escolhido no cluster. Ao ficar online, o líder zumbi irá confrontar a sua cópia do epoch number contra a do ZooKeeper, verificando que existe um novo controlador no cluster.

O que fazer para ter mais disponibilidade e consistência?

No Apache Kafka

Desabilitar o atributo "unclean.leader.election.enable"

• Caso este atributo esteja habilitado, o Apache Kafka assume a possibilidade de eleger um broker com mensagens não sincronizadas, fazendo com que essas mensagens sejam perdidas;
• Com esse atributo desativado, se a réplica líder ficar indisponível e não houver nenhuma réplica em sincronia para ser substituída, a partição ficará indisponível até que a réplica líder ou alguma réplica sincronizada esteja on-line novamente.

Replication Factor >= 3

• Para sistemas de alta disponibilidade é recomendado um fator de replicação de no mínimo 3, isso requer ao menos um cluster com 3 brokers. Isso pode ser ajustado na configuração padrão do Apache Kafka configurando o atributo default.replication.factor ou pode ser feito no momento da criação do tópico passando o parâmetro --replication-factor 3 para o arquivo kafka-topics.sh.

Garantir um número mínimo de réplicas sincronizadas configurando o atributo min.insync.replicas

• É possível configurar um número mínimo de ISRs que deverão estar disponíveis para o produtor enviar uma mensagem com sucesso para o Apache Kafka. O atributo min.insync.replicas possibilita que o produtor só receba a confirmação ack quando um número mínimo de réplicas contiver o dado;
• Se o atributo min.insync.replicas estiver configurado para 2, e se o driver do Apache Kafka do produtor estiver com o atributo acks configurado para all, cada mensagem será escrita com sucesso em ao menos duas réplicas, isso garante que a mensagem não será perdida, a não ser que ambas as réplicas sejam perdidas sem sincronizar com a terceira;
• Se o atributo min.insync.replicas estiver configurado para 2 e não houver ao menos dois brokers disponíveis, os produtores receberão o erro de LeaderNotAvailable;
• É possível configurar o min.insync.replicas por tópico, não sendo necessário fazer para o cluster inteiro.
• Se o atributo min.insync.replicas estiver configurado para o tópico e o driver do Apache Kafka do produtor não estiver com o atributo ack igual a all, o mesmo não se tornará efetivo, o Kafka dará bypass na configuração, seguindo com a configuração do produtor.

Na aplicação

Opções para os produtores de mensagens (producers)

• Esperar que cada mensagem enviada esteja em todas as réplicas de sincronização respeitando o atributo min.insync.replicas. Para isto, o ack precisar ser igual a all;
• Esperar que apenas o líder reconheça a mensagem enviada. Para isto o ack precisar estar configurado como leader;
• Não esperar o reconhecimento do Apache Kafka, a mensagem só é escrita para o buffer de rede. Para isto, o ack precisa estar configurado para none.

Para um cenário de alta consistência recomendamos configurar as propriedades do produtor para:

• ack = all
• retries = 2147483647 (max integer)
• max.block.ms = 9223372036854775807 (max long)

Cada uma das configurações descritas neste artigo possui vantagens e desvantagens e cabe avaliar para cada cenário qual a melhor opção a ser aplicada. Ou seja, o que queremos deixar claro é que não existe uma única fórmula capaz de afirmar que em um dado cenário uma configuração já utilizada em um determinado cluster se comportará da mesma forma para atender uma volumetria de processamento esperada, é necessário a realização de muitos testes, e fazer tradeoffs entre consistência e desempenho, o que inevitavelmente acarreta em muitos ajustes finos.

Opções para os consumidores (CONSUMERS)

No lado da aplicação consumidor, só podemos ler as mensagens confirmadas commited ou seja, aquelas que foram escritas para todas as ISRs.

Configurar o atributo auto.commit.enable para false

• O consumidor lê os dados da partição e confirma - commit - somente após o processamento dos dados, se por algum motivo o consumidor se tornar indisponível, ao voltar o consumidor retomará de onde parou.

Sync / Async

O commit manual pode ser feito de duas maneiras:

Síncrona

• Isto significa que a sua thread ficará bloqueada até receber uma confirmação de sucesso ou falha do Apache Kafka.

Assíncrona

• É uma confirmação não bloqueante, dessa forma a sua thread não será bloqueada. Se houver algum erro, o mesmo é passado através de um callback se fornecido, se não o erro é descartado.

Conclusão

A adoção do Apache Kafka vem crescendo rapidamente, fornecendo um conjunto de serviços escaláveis que possibilitam a integração dos dados em tempo real, com vários benefícios chaves, entre eles podemos citar:

Pode agir como um buffer para que os sistemas não travem:

• Muitos sistemas ainda fazem a transformação e o processamento de dados em lote no período da noite, o Apache Kafka resolve esse processo lento em várias etapas agindo como um intermediário recebendo os dados de um sistema de origem, e em seguida disponibilizando para o sistema de destino em tempo real.

Redução da necessidade de múltiplas integrações:

• Imagine um cenário contendo 4 sistemas de origem e 6 sistemas de destino, são necessárias a criação de 24 integrações. Esse é um processo complicado, sem mencionar também que é uma maneira lenta e propensa a erros no fornecimento de dados, com vários sistemas fazendo polling de um mesmo dado, consumindo recursos desnecessários dos nossos sistemas de gerenciamento de bancos de dados.
• Com o Apache Kafka, ao invés de construir múltiplas integrações, é necessário apenas uma integração para cada sistema produtor e consumidor, diminuindo a fricção entre times e tornando mais flexível a substituição de sistemas legados.

Baixa latência e alto throughput:

• O Apache Kafka torna mais fácil a criação de arquiteturas de sistemas em tempo real (near real time), reduz o tempo de espera pela disponibilidade de um determinado dado pela área usuária, e possibilita a tomada de decisões próximo ao tempo real, um exemplo seria o marketing de precisão.

Todos podem acessar os dados:

• Como os dados estão centralizados no Apache Kafka, qualquer outro sistema pode acessar esses dados sem impactar nos sistemas de origens. A maioria das integrações geralmente busca os dados no sistema de origem "tabela muro", podendo às vezes causar lentidão ou indisponibilidade na origem.

É essencial o conhecimento e a aplicação de abordagens de testes que estressem suficientemente os cenários passíveis de perda de dados, possibilitando a recuperação de clusters Apache Kafka sem impactar em perda de dados.

Os cenários de falha abordados neste artigo são situações que podem ocorrer em qualquer instalaçäo de Apache Kafka. Já presenciamos problemas de indisponibilidade em nós de nosso virtualizador que causaram a perda de dados em clusters não configurados apropriadamente. Hoje entendemos que para os cenários onde a perda de mensagens é inaceitável, obrigatoriamente as aplicações devem estar configuradas com o driver do produtor de mensagens para ack=all, e o tópico ter o parâmetro min.insync.replicas >= 2 para consistência e alta disponibilidade, extraindo o melhor da tecnologia e entregando o melhor valor possível para o negócio.

Referências

Distributed Consensus Reloaded: Apache ZooKeeper and Replication in Apache Kafka

A Deep Dive into Kafka Controller

Kafka as a Messaging System

Scalability of Kafka Messaging using Consumer Groups

Messaging Patterns

Apache Kafka's Distributed System Firefighter - The Controller Broker

The Internals of Apache Kafka 2.3.0

Sobre os autores