DEV Community: Kafka BR

Cotas de Cliente no Apache Kafka® 2.6.0

Fabio José — Mon, 14 Sep 2020 08:53:39 +0000

Artigo publicado no Blog oficial Kafka BR

Acesse:

https://blog.kafkabr.com/posts/kafka-260-client-quotas/

Transações no Apache Kafka®

Fabio José — Mon, 14 Sep 2020 08:51:44 +0000

Artigo publicado no Blog oficial Kafka BR.

Acesse:

https://blog.kafkabr.com/posts/transacoes

Kafka: produtores idempotentes

Fabio José — Thu, 28 May 2020 01:25:58 +0000

"...idempotência é a propriedade que algumas operações têm de poderem ser aplicadas várias vezes sem que o valor do resultado se altere após a aplicação inicial."

Através de uma configuração simples no produtor Kafka, pode-se alcançar a idempotência na produção de eventos, eliminando completamente a possibilidade de duplicatas nos tópicos.

Uma das grandes qualidades do Kafka é a possibilidade de configurar a confiabilidade no produtor, que permite ajustes finos para atender com maestria aos requisitos não-funcionais dos casos de uso.

Uma dessas possibilidades é o produtor idempotente. Com ele existe a garantia de que não haverá duplicatas em nível de partição.

Essa garantia entra em ação quando a seguinte propriedade é definida como true:

enable.idempotence=true

Ao habilitar a idempotência no produtor, também será mandatório definir as seguintes configurações:

acks=all
max.in.flight.requests.per.connection=5

Com este setup, cada requisição que envia lotes de eventos ao Kafka terá um identificador único.

Cada lote na requisição tem como destino uma partição e um tópico

Ao chegar no kafka, esta requisição será processada e cada lote enviado ao líderes das respectivas partições. Ao terminar isso, o ack é devolvido ao produtor.

Mas esta é uma comunicação em rede. Então o atraso em receber o ack ou qualquer outro problema poderá provocar a retentativa padrão dos produtores Kafka. Assim, ele enviará novamente a requisição identificada.

O broker ao receber novamente àquela requisição processada com sucesso, apenas devolve o ack, sem persistir novamente e assim evitando a duplicação.

Sem o produtor idempotente, o broker receberia a requisição da retentiva como se fosse uma nova, persistindo novamente o lote de eventos, causando duplicatas.

É isso pessoal!

Até o próximo!!

Apagar eventos do kafka?

Fabio José — Mon, 25 May 2020 16:23:28 +0000

Você já deve ter-se perguntado se é possível apagar eventos do Kafka, não? Bem, isso realmente não é possível.

Mas existe uma forma especial utilizada principalmente para atender o direito ao esquecimento, requerido pela LGPD em alguns países.

⚠️ Atenção! ⚠️ Entenda todos os impactos antes de alterar qualquer tópico que você já tenha ai no seu cluster Kafka.

Isso não se trata de um hack ou algo do tipo, até porque ele está descrito no livro "Kafka: The Definitive Guide" 😊.

Primeiro, seu tópico deverá ser compactado, ou seja, a configuração cleanup.policy deve ser igual a compact. Ele garantirá que somente o evento mais recente será mantido.

Em um tópico compactado é mandatório produzir eventos com chave de partição, o atributo key. É com base nele que o Kafka manterá o evento mais recente.

Digamos que existe o tópico clientes com os seguintes eventos na partição 3:

CHAVE   VALOR DO EVENTO
C001    Cliente 1y
C002    Cliente 2+
C003    Cliente 3@
C001    Cliente 1π

Veja que a chave C001 tem um evento mais recente, que será mantido após a compactação do tópico:

CHAVE   VALOR DO EVENTO
C002    Cliente 2+
C003    Cliente 3@
C001    Cliente 1π

Agora que você entendeu a compactação, para "apagar" um evento, o C003 por exemplo, bastará produzir um evento sem valor com a mesma chave:

CHAVE   VALOR DO EVENTO
C001    Cliente 1y
C002    Cliente 2+
C003    Cliente 3@
C001    Cliente 1π
C003

Então, após compactação do tópico, o cliente C003 deixará de existir:

CHAVE   VALOR DO EVENTO
C001    Cliente 1y
C002    Cliente 2+
C001    Cliente 1π
C003

Aqui foi um exemplo, mas a chave de partição poderia ser o CPF, CNPJ ou qualquer outra valor pertinente ao caso se uso.

Simples, não? Mas não se engane, porque empregar chave de partição não uma regra geral. Então, não mude as configurações do seu tópico antes de entender as consequências.

Se você já tem um tópico com eventos produzidos sem chave, a primeira coisa que acontecerá quando mudar a cleanup.policy para compact é que todos eles serão eliminados 😬. Mudar configurações de tópicos só com muita análise dos impactos.

É isso.

Obrigado pelo leitura e até o próximo!

Como implementar Dead-letter Topic com Spring Kafka

Fabio José — Mon, 18 May 2020 11:47:35 +0000

Dead-letter Topic, Dead-letter Queue ou em bom e velho português: Tópicos de mensagens não-entregues. São tópicos necessários em sistemas distribuídos onde a comunicação é assíncrona e através de brokers como o Kafka.

Os dados que chegam nestes tópicos passaram por todas as tentativas possíveis para tratamento de erros e já não resta mais nada a ser feito, se não, a intervenção humana. Assim, não será qualquer erro que levará mensagens ou eventos a serem publicados em um tópico dead-letter.

No mundo Kafka um tópico dead-letter é destinado aos registros consumidos, que por algum erro irrecuperável não puderam ser processados com sucesso.

Neste artigo será demonstrada uma abordagem para implementar DLT no Kafka, utilizando Java e Spring. Para os impacientes 🧐, estes são os fontes :

fabiojose / spring-kafka-dlt-ex

Dead-letter Topic com Spring Kafka

Classificar Erros

Antes de escrever qualquer linha de código é necessário classificar os erros e como serão tratados.

Existem dois tipos de erros:

recuperáveis
não-recuperáveis

Os erros recuperáveis ou que podem ser tratados, são aqueles onde alguma abordagem será empregada para tentar finalizar o fluxo com sucesso.

Por exemplo, no Java, os erros ConnectException e UnknownHostException podem ser recuperados através de retentativas, pois normalmente são causados por instabilidades momentâneas no serviço ou na rede.

Já os não-recuperáveis são aqueles que independente do que seja feito, não será possível finalizar o fluxo com sucesso. Logo, estes são candidatos a seguirem diretamente para o tópico dead-letter. E como exemplo, se você utiliza Avro, o erro AvroMissingFieldException indica a ausência de um campo requerido, não havendo nada a fazer. Portanto será inútil a retentativa, por exemplo.

E além dos erros técnicos, você também deverá classificar seus erros de negócio e escolher uma estratégia para tratá-los.

Bem, agora veremos como implementar DLT para problemas técnicos no Java. E é bem provável que você encontre equivalentes na sua linguagem ou framework.

Recuperáveis

Aqui segue uma lista de erros técnicos recuperáveis que, em nome da simplicidade, serão tratados através de retentativas.

ConnectException
UnknownHostException
ProductNotFoundException: a título de exemplo, este erro de negócio foi definido como tratável. Porque um serviço chamado Catalogo, que faz event-sourcing das mudanças emitidas pelo serviço Produto, ainda não processou o evento com o produto em questão. Mas através da retentativa seu fluxo poderá finalizar com sucesso.

Não-recuperáveis

Estes são alguns dos erros não-recuperáveis, ou seja, se passarem pelo mesmo processo de retentativas somente consumiriam recursos, sem chances de finalizar com sucesso.

AvroMissingFieldException
NullPointerException
ClassCastException
RecordTooLargeException
StockNotAvailableException: como exemplo, este erro de negócio foi classificado como irrecuperável. Porque a falta de produtos no estoque não será resolvida com retentativas, por exemplo.

Implementação

Existe um ótimo artigo no blog de engenharia do Uber que descreve uma arquitetura para dead-letter. Vale a leitura!

Esta implementação foi dividida em três partes:

tópicos
construção
tratamento

Tópicos

Para cada processo que terá tratamento dead-letter, será criado um tópico com o sufixo -dlt. Tome como exemplo um processo que realiza a reserva de estoque a partir das ordens de compra publicadas no tópico ordem-compra. Então, ao invés de criar um tópico ordem-compra-dlq, será utilizado um nome relevante ao processo que está falhando:

reservar-estoque-dlt

E mais os tópicos para retentativas, que devem ser quantos forem necessários até finalmente o registro chegar ao tópico dead-letter. Digamos que serão no máximo quatro retentativas além da inicial:

reservar-estoque-retry-1
reservar-estoque-retry-2
reservar-estoque-retry-3
reservar-estoque-retry-4

Os tópicos devem ter características que não interfiram no andamento das retentativas ou na publicação do tópico dead-letter. Um exemplo é a configuração max.message.bytes, que pode acarretar um erro chamado message too large. Portanto os tópicos deverão permitir mensagens maiores, caso contrário também não será possível utilizá-los porque são idênticos ao principal, inclusive com as mesmas limitações.

Construção

A construção foi feita com Spring Kafka porque ele já vem com muitas utilidades para tratamento para dead-letter, o que contribui muito com a produtividade se ela é o foco no seu projeto. Mas nada impede que você escreva a mesma solução com Clientes Kafka no Java ou na sua linguagem do seu projeto.

Para utilizar os recursos destinados a dead-letter, é necessário customizar a fábrica de objetos encarregada de produzir listeners Kafka no Spring.

Assim, a configuração programática foi sub-dividida em três partes:

resolver: responsável por determinar o tópico destino do registro que está no contexto do erro.
errorHandler: manipulador do erro
kafkaListernerContainerFactory: fabrica instâncias que são utilizadas nos métodos anotados com @KafkaListener

E para cada uma das sub-divisões existe uma implementação Main e outra Retry. Como é possível imaginar, uma cuida das configurações para o processamento principal outro para as retentativas.

Main resolver, responsável por determinar qual o tópico destino com base no erro-raiz lançado ao processar o registro consumido do tópico ordem-compra:



@Bean
public BiFunction < ConsumerRecord < ? , ? > , Exception, TopicPartition >
 mainResolver() {

  return new BiFunction < ConsumerRecord < ? , ? > , Exception, TopicPartition > () {
   @Override
   public TopicPartition apply(ConsumerRecord < ? , ? > r, Exception e) {

    // ####
    // Por padrão, quando é não-recuperável, segue diretamente p/ dead-letter
    TopicPartition result =
     new TopicPartition(dltTopic,
      QUALQUER_PARTICAO);

    // ####
    // Trata-se de um erro recuperável?
    final boolean recuperavel = isRecuperavel(e);
    if (recuperavel) {

     Optional < String > origem =
      topicoOrigem(r.headers())
      .or(() -> Optional.of(NENHUM_CABECALHO));

     // ####
     // Se origem for outro tópico, segue para o primeiro retry
     String destino =
      origem
      .filter(topico -> !topico.matches(retryTopicsPattern))
      .map(t -> retryFirstTopic)
      .orElse(dltTopic);

     result = new TopicPartition(destino, QUALQUER_PARTICAO);
    }

    return result;
   }
  };
 }

Main error handler, que utiliza o Main resolver e é responsável por definir duas configurações essenciais para tratamento dos erros:

DeadLetterPublishingRecoverer: inicia o fluxo DLT, que é delegado pelo SeekToCurrentErrorHandler caso as retentativas locais não resolvam o erro.
SeekToCurrentErrorHandler: manipula qualquer erro que seja lançado no método anotado com @KafkaListener. Naturalmente, se você capturá-los com catch e não permitir que eles subam na pilha, não será possível tratá-los.



@Bean
public SeekToCurrentErrorHandler mainErrorHandler(
 @Qualifier("mainResolver")
 BiFunction < ConsumerRecord < ? , ? > , Exception, TopicPartition > resolver,
 KafkaTemplate < ? , ? > template) {

 // ####
 // Recuperação usando dead-letter 
 DeadLetterPublishingRecoverer recoverer =
  new DeadLetterPublishingRecoverer(template, resolver);

 // ####
 // Tentar 3x localmente antes de iniciar o fluxo dead-letter
 SeekToCurrentErrorHandler handler =
  new SeekToCurrentErrorHandler(recoverer, RETENTAR_3X);

 // ####
 // Lista das exceções não-recuperáveis, para evitar o retry local
 excecoes.getNaoRecuperavies().forEach(e ->
  handler.addNotRetryableException(e));

 return handler;
}

Main listener factory, encarregado de fabricar os consumidores que processam registros do tópíco ordem-compra.



@Bean
public KafkaListenerContainerFactory < ConcurrentMessageListenerContainer < String, GenericRecord >>
 mainKafkaListenerContainerFactory(
  @Qualifier("mainErrorHandler") SeekToCurrentErrorHandler errorHandler,
  KafkaProperties properties,
  ConsumerFactory < String, GenericRecord > factory) {

  ConcurrentKafkaListenerContainerFactory < String, GenericRecord > listener =
   new ConcurrentKafkaListenerContainerFactory < > ();

  listener.setConsumerFactory(factory);

  // ####
  // Utilizando o mainErrorHandler para tratar os erros
  listener.setErrorHandler(errorHandler);

  // Falhar, caso os tópicos não existam?
  listener.getContainerProperties()
   .setMissingTopicsFatal(missingTopicsFatal);

  // Commit do offset no registro, logo após processá-lo no listener
  listener.getContainerProperties().setAckMode(AckMode.MANUAL);

  // Commits síncronos
  listener.getContainerProperties().setSyncCommits(Boolean.TRUE);

  return listener;
 }

Então, basta anotar o método como segue.

Mas vale ressaltar que o offset sempre deverá ser confirmado, porque todo o fluxo que trata os erros será feito por retentativa e talvez culminando no tópico dead-letter. Analise o seu caso-de-uso e entenda se esta abordagem também se aplicada.



@KafkaListener(
 id = "main-kafka-listener",
 topics = "${app.kafka.consumer.topics}",
 containerFactory = "mainKafkaListenerContainerFactory"
)
public void consume(@Payload ConsumerRecord < String, GenericRecord > record,
 Acknowledgment ack) throws Exception {

 try {

  // #### 
  // Processar
  process(record);

 } finally {

  // ####
  // Sempre confirmar o offset
  ack.acknowledge();
 }

}

Já a configuração para o processamento das retentativas segue moldes similares, com algumas exceções:

resolver: retorna qual o próximo tópico na sequência de retentativas ou se é o reservar-estoque-dlt, caso já tenham passadas por todas.
errorHandler: nenhuma retentiva local.

Este é o método anotado com @KafkaListener que tratará os tópicos reservar-estoque-retry:



@KafkaListener(
 id = "retry-kafka-listener",
 topicPattern = "${app.kafka.dlt.retry.topics.pattern}",
 containerFactory = "retryKafkaListenerContainerFactory",
 properties = {
  "fetch.min.bytes=${app.kafka.dlt.retry.min.bytes}",
  "fetch.max.wait.ms=${app.kafka.dlt.retry.max.wait.ms}"
 }
)
public void retry(@Payload ConsumerRecord < String, GenericRecord > record,
 Acknowledgment ack) throws Exception {

 try {

  // ####
  // Reprocessar
  process(record);

 } finally {
  // ####
  // Sempre confirmar o offset
  ack.acknowledge();
 }

}

topicPattern: consumir todos os tópicos reservar-estoque-retry
fetch.min.bytes e fetch.max.wait.ms: utilizados para provocar um certo atraso no consumo dos registros, visto que sem eles o consumo seria praticamente instantâneo.

Por fim, o application.properties será assim:



app.kafka.dlt.retry.topics=4
app.kafka.dlt.retry.topics.pattern=reservar-estoque-retry-[0-9]+
app.kafka.dlt.retry.topic.first=reservar-estoque-retry-1
app.kafka.dlt.topic=reservar-estoque-dlt

# Lista de exceções recuperáveis
app.kafka.dlt.excecoes.recuperaveis[0]=java.net.ConnectException
app.kafka.dlt.excecoes.recuperaveis[1]=java.net.UnknownHostException

# Lista de exceções não-recuperáveis
app.kafka.dlt.excecoes.naoRecuperaveis[0]=org.apache.avro.AvroMissingFieldException
app.kafka.dlt.excecoes.naoRecuperaveis[1]=java.lang.NullPointerException

# Provocar atraso no processmento de retentativa
app.kafka.dlt.retry.max.wait.ms=20000
app.kafka.dlt.retry.min.bytes=52428800

Todo a implementação está disponível no Github, clone it and have fun!

fabiojose / spring-kafka-dlt-ex

Dead-letter Topic com Spring Kafka

spring-kafka-dlt-ex

Dead-letter Topico com Spring Kafka e formato de dados Avro.

Requerimentos

JDK 11
Apache Maven 3.6+
Docker 19+
Acesso ao repositório https://repo.maven.apache.org/maven2/ ou uma alternativa com acesso às dependências presentes no pom.xml
Schema Registry
Kafka
Testcontainers
Lombok
Commons Lang3

Configurações

Não se preocupe, pois apesar de existirem atalhos pelas variávies de ambiente, você pode utilizar tranquilamente aquilo que o Spring Boot oferece. Então veja todos as propriedades no application.properties

No caso do Kafka, utilizamos Spring Kafka, então você utilizar o modo Spring para configurações.

Variáveis de Ambiente

APP_KAFKA_CONSUMER_TOPICS: tópicos para consumir, ou expressão.
KAFKA_CLIENT_ID: nome do cliente Kafka, usado pelos brokers para logs e métricas. Utilize um nome clean, não genérico.
- spring.kafka.producer.client-id, spring.kafka.consumer.client-id
KAFKA_CONSUMER_GROUP: nome do grupo de consumo que esta aplicação pertence
- spring.kafka.consumer.group-id
KAFKA_BOOTSTRAP_SERVERS: lista de brokers para o cluster Kafka
- spring.kafka.bootstrap-servers
SCHEMA_REGISTRY_URL: url para o registro de esquemas Avro
- spring.kafka.properties.schema.registry.url
…

View on GitHub

Quando o registro atinge o último tópico de retentativa, que neste caso é o reservar-estoque-4 e não seja processado com sucesso, finalmente ele será publicado no tópico dead-letter, então a equipe deverá estar preparada para o tratamento adequado.

Tratamento

Bem, neste momento o registro com problemas já desembarcou no tópico reservar-estoque-dlt e bem antes disso acontecer um sistema preciso de monitoramento dos tópicos deveria ter alertado sobre o uso dos tópicos para retentativas, principalmente se os registros estão atingindo o reservar-estoque-retry-4.

Veja neste artigo como monitorar seu cluster Kafka.

A maneira mais prudente de tratamento, além do monitoramento e um ótimo sistema de rastreamento distribuído, será construir um processo em que para cada registro publicado no DLT, tickets e notificações ChatOps deveram ser enviados para a equipe responsável.

Também são necessárias ferramentas adequadas para, por exemplo, editar registros e colocá-los novamente no tópico original.

Bem, é isso! Até o próximo.

Kafka: consumindo registros com Spring

Fabio José — Tue, 28 Apr 2020 11:48:36 +0000

O consumo de dados do Kafka, dependendo do caso de uso, requer alguns cuidados no commit. Fazê-lo de forma correta é importante para garantir que nada seja perdido.

Com Clientes Kafka, em especial a versão para Java, é muito simples e intuitivo, mas no Spring Kafka é necessária atenção com ajustes especiais.

O foco neste artigo é mostrar como consumir dados com Spring Kafka, que é uma abstração sobre os Clientes Kafka para Java. Fazendo isso com seguindo setup:

enable.auto.commit=false. Que requer commit manual
commit síncrono

Devido a semântica de entrega at-least-once, um registro poderá ser consumido uma ou n vezes, e este setup busca reduzir isso.

Um consumidor típico no Spring Kafka é escrito assim:



@Component
public class SpringKafkaListener {
  @KafkaListener(topics = "topico")
  public void consume(String valor) {
    // Processar valor do registro
  }
}

E criado com as seguintes configurações, feitas no application.properties:



spring.kafka.bootstrap-servers=configure-me_kafka-broker:9092
spring.kafka.consumer.client-id=configure-me_client-id
spring.kafka.consumer.group-id=configure-me_group-id
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

Felizmente o Spring Kafka não redefine valores, portanto a configuração padrão é mantida assim como definida na documentação oficial, porém, nela o commit é automático e assíncrono. Contudo, no Spring Kafka, todas as configurações necessárias para commit manual e síncrono não estão disponíveis através de propriedades no application.properties.

Resolvendo

Spring Kafka é uma abstração, logo o poll loop e commit são transparentes. E como pode-se ver no exemplo, um consumidor recebe apenas o registro e por padrão não tem acesso ao Consumer.

Primeiro é necessário revisar as configurações para desligar o commit automático.

Nova configuração:



# Nada de novo aqui
spring.kafka.bootstrap-servers=configure-me_kafka-broker:9092
spring.kafka.consumer.client-id=configure-me_client-id
spring.kafka.consumer.group-id=configure-me_group-id
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

# Desliga o commit automático no Cliente Kafka
spring.kafka.consumer.enable-auto-commit=false

Spring tem sua própria notação para a maioria das configurações presentes no Kafka Consumer, que são traduzidas em tempo de execução para o nome correto.

Agora que o commit automático foi desligado, são necessários alguns ajustes programáticos feitos ao customizar as fábricas de objetos:

ackMode para MANUAL
syncCommits como true



import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties.AckMode;

@EnableKafka
@Configuration
public class KafkaConfig {

  @Autowired
  KafkaProperties properties;

  @Bean
  public ConsumerFactory<String, String> consumerFactory() {
      return new DefaultKafkaConsumerFactory<>(
              properties.buildConsumerProperties());
  }

  @Bean
  public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>>
      kafkaListenerContainerFactory() {

      ConcurrentKafkaListenerContainerFactory<String, String> listener = 
            new ConcurrentKafkaListenerContainerFactory<>();

      listener.setConsumerFactory(consumerFactory());

      // Não falhar, caso ainda não existam os tópicos para consumo
      listener.getContainerProperties()
          .setMissingTopicsFatal(false);

      // ### AQUI
      // Commit manual do offset
      listener.getContainerProperties().setAckMode(AckMode.MANUAL);

      // ### AQUI
      // Commits síncronos
      listener.getContainerProperties().setSyncCommits(Boolean.TRUE);

      return listener;
    }
}

Então o consumidor com Spring Kafka terá esta aparência:



import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

@Component
public class SpringKafkaListener {

  @KafkaListener(topics = "topico")
  public void consume(@Payload String valor, Acknowledgment ack) {

    //TODO Processar registro
    // . . . 

    // Commmit manual, que também será síncrono
    ack.acknowledge();

  }
}

Note que mesmo assim não existe acesso ao Consumer, ao invez disso, o Spring injeta uma instância de Acknowledgment que faz o commit quando tem seu método acknowledge() executado.

Também neste exemplo o offset é confirmado a cada registro processado. Isso é algo que degrada a taxa de transferência, mas reduz ainda mais as chances de consumos duplicados. Bem, mas cada caso é um caso 😊.

O exemplo completo está disponível no Github:

fabiojose / skc-ex

Spring Kafka Consumer

Spring Kafka Consumer Example

Exemplo de consumer com Spring Kafka

Requerimentos

JDK 1.8
Acesso ao repositório https://repo.maven.apache.org/maven2/ ou uma alternativa com acesso às dependências presentes no pom.xml

Build & Run

Maven

Para montar o fatjar, execute o comando:

Linux

./mvnw clean package

Windows

.\mvnw.cmd clean package

Para executar:

Você pode utilizar o docker-compose.yaml para subir um Kafka em sua máquina

java \
  -Dspring.kafka.bootstrap-servers='localhost:9092' \
  -Dspring.kafka.consumer.client-id='spring-kafka-ex' \
  -Dspring.kafka.consumer.group-id='meu-grupo' \
  -jar target/app-spring-boot.jar

Docker

A definição Dockerfile desta aplicação emprega multi-stage builds. Isso significa que nela acontece o build da aplicação e a criação da imagem.

Se for necessário somente a criar a imagem, pode-se utilizar a definição Dockerfile-image. Mas antes é necessário montar o fatjar através do maven.

Para build do fatjar e montar a imagem, execute o comando:

docker build . -t sk-consumer-ex:1.0

Para montar apenas a imagem (antes…

View on GitHub

Photo by Paweł Czerwiński on Unsplash

Kafka: compactação e compressão

Fabio José — Mon, 20 Apr 2020 14:58:36 +0000

Compactar e comprimir tem algo em comum:

reduzir o espaço ocupado pelos dados

Seja removendo partes desnecessárias através da compactação ou tornado-as menores com algoritmos de compressão.

No Kafka podemos utilizar ambas.

Compactação

A compactação no Kafka pode ser ativada através de uma configuração feita nos tópicos:

cleanup.policy=compact

Por padrão ela é definida como delete. Veja outros detalhes na documentação oficial.

Quando a política de limpeza for igual a compact, periodicamente os tópicos são varridos em busca de registros repetidos. E para encontrá-los, o Kafka utiliza a key presente em cada um deles.

Registros sem chave, são imediatamente eliminados.

Acompanhe este exemplo, onde os registros estão dispotos dos mais antigos para os mais novos;

Key  Value
c1   Graziela Maciel, MS
c3   Carlos Chagas, MG
c5   Bertha Lutz, SP
c1   Graziela Maciel, RJ
c3   Carlos Chagas, RJ
c5   Bertha Luttz, SP
c7   Suzana Herculano, RJ
c5   Bertha Lutz, RJ

Estes são os registros mantidos após o processo de compactação:

Key  Value
c1   Graziela Maciel, RJ
c3   Carlos Chagas, RJ
c7   Suzana Herculano, RJ
c5   Bertha Lutz, RJ

É assim que funciona a compactação no Kafka, somente o registro mais recente é mantido. Todos os outros mais antigos e com a mesma key são eliminados.

Por exemplo, existem três registros com a chave c5 e o mais recente é Bertha Lutz, RJ, que foi mantido pela compactação. Note que haviam 8 registros, restando apenas 4.

Compressão

A compressão pode ser feita através de uma configuração nos tópicos ou nos produtores.

No tópico a compressão é ativada através da configuração compression.type, que por padrão está definida como producer. Isso significa que a compressão empregada pelos produtores será mantida e repassada aos consumidores, mesmo que não exista compressão alguma.

Os valores possíveis para compression.type são:

uncompressed: mesmo que os produtores enviem dados comprimidos, eles serão persistidos no seu tamanho original, ou seja, sem compressão.
zstd: algoritmo de compressão criado pelo Facebook.
lz4: a compressão mais recomendada pelos especialistas.
snappy: algoritmo criado pelo Google.
gzip: ótimos níveis de compressão, usando com mais intensidade a cpu.
producer: mantém a compressão do produtor, se existir.

E no produtor também existe a configuração compression.type, que possuí os mesmos valores, com exceção de uncompressed e producer. Adicionalmente, existe a opção none para não aplicar compressão, que também é o valor padrão.

Comprimir dados no produtor é interessante porque reduz o uso da banda de rede e de espaço em disco e cpu nos brokers.

E a compressão traz bons resultados quando são produzidos lotes de dados, pois ela não é aplicada em nível de registro. Ou seja, tanto no produtor, quando no Kafka, a compressão não é em cima de cada mensagem ou evento.

Já nos consumidores, nenhuma configuração especial é necessária, porque através do protocolo Kafka ele verifica se há compressão e emprega o mesmo algoritmo.

Kafka: Garantia de Ordem

Fabio José — Sat, 11 Apr 2020 11:09:24 +0000

Existem casos de uso onde a manutenção da ordem em que os dados foram produzidos é um requisito. Há outros em que isso não importa.

Event Sourcing é um exemplo onde a manutenção da ordem importa, caso contrário o estado final seria inconsistente.

Kafka, através das configurações corretas, pode nos ajudar em qualquer um deles.

Como é a ordenação no Kafka

A ordem é garantida em nível de partição, ou seja, tópicos com várias partições não possuem ordenação global dos dados.

Bem, dado um lote de registros produzidos pelo produtor p1 que tem como destino a partição 4 no tópico t1, haverá garantia de que a ordem produzida será mantida pelo Kafka e repassada aos consumidores.

Desse modo se houver um produtor p2 também produzindo lotes de registros no mesmo destino de p1, nenhuma verificação de ordem será realizada entre os lotes destes produtores, prevalecendo àquele que primeiro chegar.

Mas existem situações e configurações que podem interferir e mudar a ordenação dos registros criados por um determinado produtor p.

O que pode mudar a ordenação

Há uma dezena de configurações que podem ser ajustadas para determinar como será o comportamento do nosso produtor. E muitas já possuem valores padrão que atendem alguns casos de uso, como:

retries: 2¹⁶
- número de retentativas
max.in.flight.requests.per.connection: 5
- requisições que ainda não receberam ack, ou seja, estão em voo.
acks: 1
- reconhecimento de recebimento e não

O valor padrão de retentativas, definido por retries, é bem grande e elas são realizadas para todos os problemas relativamente recuperáveis, como uma falha na rede.

E o número de requisições em voo é igual a 5. E isso poderá causar a troca de ordem.

Digamos que se produz o primeiro lote de registros, l1, que fica em voo aguardando reconhecimento. Em seguida o mesmo produtor envia o segundo lote, l2. Pela ordenação, l1 contém registros que estão antes daqueles em l2.

Essa é a ordem criada pelo produtor:

l1->1,2,3,4
l2->5,6,7

Espera-se que seja mantida:

1,2,3,4,5,6,7

Porém, se l1 for para retentativa, l2 for efetivado e l1 efetivado em seguida, a ordenação muda, não prevalecendo àquela criada no produtor.

Realidade como descrito acima:

5,6,7,1,2,3,4

Configuração correta

Felizmente é possível adequar as configurações do produtor e eliminar as chances da perda de ordem:

max.in.flight.requests.per.connection deverá ser configurado para que apenas uma requisição fique em voo, ou seja, 1.

Isso certamente irá diminuir a taxa de transferência, não de maneira drástica. Mas agora a ordenação está garantida, mesmo em situações de retentativas.

Monitorando Consumer Lag do Apache Kafka

Álvaro Bacelar — Sun, 29 Mar 2020 20:07:05 +0000

O Apache Kafka tem milhares de métricas que podem ser acessadas via interface JMX (nesse artigo eu mostro como instrumentar tais métricas utilizando ferramentas Open Source). Contudo há uma métrica, não menos importante, que não está disponível via JMX: O Consumer Lag.

Mensurada em número de mensagem, basicamente o Lag é a diferença entre a última mensagem produzida em uma partição especifica e a última mensagem processada (committed) pelo consumidor.

Segundo o livro "kafka - The Definitive Guide" o método preferido para monitorar o Lag é através de uma aplicação externa que possa visualizar o estado de uma partição no broker, rastreando o offset mais recente da mensagem produzida e o último offset processado pelo consumidor, atraves do topico interno de controle: __consumer_offsets

Levando isso em consideração, o LinkedIn (empresa responsável pela criação do Apache Kafka) desenvolveu (também) uma aplicação que realiza essa rastreabilidade dos offsets das partições dos brokers. Essa aplicação é o Burrow.

O Burrow é uma ferramenta que foi escrita na linguagem Go e ela nos provê tanto informações de Consumer Lag quanto informações diversas do cluster e disponibiliza tais valores via REST API. Com o Burrow ainda é possível enviar notificações via email ou WebHook com um threshold definido.

Como falei acima, o Burrow provê essas informações via REST API e nosso objetivo aqui é fazer essa informação chegar no Prometheus e consequentemente no Grafana.

Então irei mostrar como instalar e configurar o Burrow juntamente com o burrow_exporter (também escrito em Go) para monitorar o Consumer Lag de um cluster Apache Kafka e ter todas as informações do Lag, entre outras, no Prometheus.

Para realizar todos os testes desse artigo, eu realizei o setup de um cluster Apache Kafka com o Ansible. Clicando aqui você verá um artigo que escrevi mostrando como realizar um setup de um cluster Apache Kafka e Zookeeper com o Ansible.

Instalando e Configurando o Prometheus

Para instalar e configurar o Prometheus vamos seguir os seguintes passos:

Baixar a versão mais estável do Prometheus ```

wget https://github.com/prometheus/prometheus/releases/download/v2.12.0/prometheus-2.12.0.linux-amd64.tar.gz

- Descompactando no diretório /srv

tar -zxvf prometheus-2.12.0.linux-amd64.tar.gz -C /srv/

- Adicionar usuário de serviço do prometheus

useradd -s /usr/sbin/nologin prometheus

- Mudar o dono da pasta do prometheus

chown prometheus:prometheus /srv/prometheus-2.12.0.linux-amd64/ -R

- Adicionar o arquivo de service do prometheus

vim /etc/systemd/system/prometheus.service

[Unit]
Description=Prometheus
After=network-online.target
[Service]
Type=simple
User=prometheus
Group=prometheus
ExecReload=/bin/kill -HUP $MAINPID
ExecStart=/srv/prometheus-2.12.0.linux-amd64/prometheus --config.file=/srv/prometheus-2.12.0.linux-amd64/prometheus.yml --storage.tsdb.path=/srv/prometheus-2.12.0.linux-amd64/data --web.listen-address=0.0.0.0:9090
LimitNOFILE=65000
LockPersonality=true
NoNewPrivileges=true
MemoryDenyWriteExecute=true
PrivateDevices=true
PrivateTmp=true
ProtectHome=true
RemoveIPC=true
RestrictSUIDSGID=true
ProtectSystem=full
SyslogIdentifier=prometheus
Restart=always
[Install]
WantedBy=multi-user.targe

- Executar o daemon-reload e iniciar o serviço

systemctl daemon-reload

systemctl start prometheus


Se tudo tiver dado certo o serviço subiu na porta 9090 e podemos acessar nessa porta: 
http://127.0.0.1:9090

### Instalando o Grafana

A instalação do Grafana é mais fácil, para instala-lo basta acessar o seguinte link https://grafana.com/grafana/download e seguir os passos descrito no link de acordo com seu S.O. 

Depois de instalado suba o serviço

systemctl start grafana-server


O Grafana por padrão sobe na porta 3000, então vamos acessa-lo:
http://127.0.0.1:3000

O usuário e senha default do Grafana é admin, no seu primeiro acesso é solicitado para trocar.

Após acessar o Grafana, você deve adicionar o datasource do Prometheus.

### Instalando e configurando o Burrow

O primeiro passo para instalar o Burrow é realizar o download do código fonte, para isso vá ao repositório oficial do Burrow no link abaixo:

$ git clone https://github.com/linkedin/Burrow.git


Com o código fonte do Burrow, vamos entrar no diretório e *buildar* a imagem Docker:

$ cd ${PWD}/Burrow
$ docker build -t burrow-api .




>Nesse post irei focar na instalação do Burrow via [Docker](https://www.docker.com/). Para aqueles que preferirem executar o Burrow sem o uso do Docker, é só seguir os passos que estão descritos no [README.rd](https://github.com/linkedin/Burrow#build-and-install) do repositório do Burrow.

Com a imagem do Burrow *buildada* vamos agora realizar o mesmo procedimento com a imagem do burrow_exporter. Baixe o código fonte no link abaixo: 


  
    
      
      
        alvarobacelar
       / 
        burrow_exporter
      
    
    
      A Prometheus Exporter for gathering Kafka consumer group info from Burrow

git clone https://github.com/alvarobacelar/burrow_exporter.git


> O repositório acima é um fork do projeto [burrow_exporter](https://github.com/shamil/burrow_exporter) criado por Shamil. Eu solicitei um [Pull Request](https://github.com/shamil/burrow_exporter/pull/1) há um tempo para adicionar alguns recursos extras. Mas o dono do repositório nunca respondeu, então vamos seguir usando o projeto que *forkei*, pois ele retorna um valor a mais que a API do Burrow nos disponibiliza (iremos ver isso logo mais na frente).

Entrando na pasta do do burrow_exporter, que acabamos de baixar, vamos *buildar* a imagem Docker:

$ docker build -t burrow-exporter .


Agora que temos as imagens do Burrow e burrow_exporter criadas, vamos 
criar o arquivo docker-compose.yml para subirmos as duas aplicações posteriormente. O arquivo deve conter o seguinte conteúdo:
```yaml


version: "2"
services:
  burrow:
    image: burrow-api
    volumes:
      - ${PWD}/config:/etc/burrow/
    container_name: burrow_api

  burrow_exporter:
    image: burrow-exporter
    container_name: burrow_exporter
    ports:
      - 8090:8237
    depends_on:
      - burrow
    command: --burrow.address http://burrow:8000

Antes de subir os containers precisamos adicionar os servidores do kafka e zookeeper no arquivo de configuração do Burrow. Dentro do diretório da API do Burrow vamos editar o arquivo burrow.toml



$ vim ${PWD}/Burrow/config/burrow.toml

No arquivo vamos alterar os endereços dos brokers, Zookeeper e outros parâmetros:



[general]
pidfile="burrow.pid"
stdout-logfile="burrow.out"
access-control-allow-origin="mysite.example.com"

[logging]
filename="logs/burrow.log"
level="info"
maxsize=100
maxbackups=30
maxage=10
use-localtime=false
use-compression=true

[zookeeper]
servers=[ "zkhost01.example.com:2181", "zkhost02.example.com:2181", "zkhost03.example.com:2181" ] # altere para os edereços de seus zookeepers
timeout=6
root-path="/burrow"

############################
####### CLUSTER zoom #######
############################
# altere o nome do profile do cliente caso queira
[client-profile.post]
client-id="burrow-monitor"
kafka-version="2.3.0" # altere para a versão 2.3.0 do Kafka

# Dê um nome para o seu cluster [cluster.nome]
[cluster.zoom]
class-name="kafka"
servers=[ "kafka01.example.com:10251", "kafka02.example.com:10251", "kafka03.example.com:10251" ] # altere para os endereços dos seus brokers
client-profile="post" # coloque aqui o nome do client-profile definido acima 
topic-refresh=120
offset-refresh=30

# Dê um nome para o seu cluster [cluster.nome]
[consumer.zoom]
class-name="kafka"
cluster="zoom" # coloque aqui o nome do cluster definido
servers=[ "kafka01.example.com:10251", "kafka02.example.com:10251", "kafka03.example.com:10251" ] # altere para os endereços dos seus brokers
client-profile="post" # coloque aqui o nome do client-profile definido 
group-blacklist="^(console-consumer-|python-kafka-consumer-|quick-).*$" # coloque aqui os nomes dos consumer groups que não quer que apareça
group-whitelist=""
################################
####### FIM CLUSTER zoom #######
################################
# Repita isso para quantos clusters tiver

[httpserver.default]
address=":8000"

[storage.default]
class-name="inmemory"
workers=20
intervals=15
expire-group=604800
min-distance=1

## Vamos deixar comentado por enquanto
# [notifier.default]
# class-name="http"
# url-open="http://someservice.example.com:1467/v1/event"
# interval=60
# timeout=5
# keepalive=30
# extras={ api_key="REDACTED", app="burrow", tier="STG", fabric="mydc" }
# template-open="conf/default-http-post.tmpl"
# template-close="conf/default-http-delete.tmpl"
# method-close="DELETE"
# send-close=true
# threshold=1

Com o arquivo docker-compose.yml criado e o arquivo de configuração burrow.toml alterado com os endereços de seus clusters, podemos subir as duas aplicações:



$ docker-compose -f docker-compose up -d

Agora que temos os dois containers rodando, vamos configurar o Prometheus para que este capture as métricas.

No arquivo de configuração do Prometheus (vim /srv/prometheus-2.12.0.linux-amd64/prometheus.yml), você deve adicionar as seguintes linhas:



- job_name: 'burrow'
    static_configs:
    - targets: ['127.0.0.1:8090']

O IP 127.0.0.1 deve ser substituído pelo o IP do servidor que está executando o container que subimos acima.

Para o Prometheus reconhecer tais configurações precisamos realizar um reload no serviço:



# systemctl reload prometheus

Agora vamos importar o dashboard no Grafana. O link abaixo é o dash específico para essas métricas:
https://grafana.com/grafana/dashboards/11963

Quando tiver importando o dash acima e tudo tiver ocorrido bem você verá algo parecido com imagem abaixo:

Lembram que eu havia comentado acima que iriamos utilizar o repositório que eu forkei? Pois bem, se você analisar o dash que importamos vai ver que tem um campo na tabela chamado Consumer client. Foi esse campo que adicionei no PR que abri e o dono do repositório nunca aceitou. Com essa informação sabemos qual é o IP do servidor que está consumindo naquela especifica partição.

Conclusão

O sucesso para que o seu cluster Apache Kafka (ou qualquer outra aplicação) esteja 100% disponível e seja 100% confiável é ter uma stack madura de monitoração. Como mostrado nesse artigo, não é necessário gastar uma bala com licenças em softwares de monitoração achando que tal software vai fazer mágica para você. Com ferramentas Open Source somos capaz de tirar o melhor que cada uma tem a oferecer e montar uma stack muito madura de monitoração e alertas.

Kafka: quantas partições por tópico?

Fabio José — Sun, 01 Mar 2020 19:19:22 +0000

Umas das grandes dúvidas ao utilizar o Kafka é saber quantas partições são necessárias ao criar novos tópicos. Bem, não existe uma fórmula geral, o que temos é uma aproximação detalhada neste excelente artigo How to choose the number of topics/partitions in a Kafka cluster?, escrito por Jun Rao. Outro artigo muito relevante é o Benchmarking Apache Kafka: 2 Million Writes Per Second, escrito por Jay Kreps. Nele são revelados resultados importantes sobre a taxa de transferência para producers e consumers.

Então, com base no artigo de Jun Rao, temos a fórmula aproximada para determinar o número de partições:

  MAX(t/p, t/c)

Onde:

t: taxa de transferência desejada
p: taxa de transferência do producer
c: taxa de transferência do consumer

Como o artigo sugere, o valor para a taxa de transferência do consumer depende de como ele processa os registros e por esse motivo devemos realizar nossas próprias medições ao invés de utilizar o valor-base descrito no artigo de Jay Kreps. Já para o valor referente ao producer, podemos tomar como base àquele revelado pelo artigo.

Uma dica é você realizar todas as medições, assim você também entenderá como é a sua infra kafka.

Aplicando a fórmula

Primeiro temos de definir qual é a unidade da nossa taxa de transferência, que poderia ser MB/s ou Mensagens/s. Mas as mensagens tem tamanhos variados e utilizá-las nas medições talvez não conduza a resultados realistas, então, MB/s é uma boa unidade de medida para aplicação da fórmula.

Unidade: MB/s (megabytes por segundo)

A medição é realizada empregando um producer e um tópico com apenas uma partição. Então, digamos que o resultado para nosso p, foi:

89 MB/s

A medição para taxa de transferência do consumer é similar, ou seja, apenas um tópico com uma única partição. Então, digamos que o valor de c é:

75 MB/s

Imagine que nosso alvo com relação a taxa de transferência seja 450 MB/s. Aplicando a fórmula de aproximação temos:

p = 89
c = 75
t = 450

  MAX(450/89, 450/75)

  MAX(5.1, 6) = 6

Portanto, o número de partições é 6.

Notadamente, não trata-se de uma fórmula para qualquer caso de uso, porém, agora temos um ponto de partida e não somente números mágicos.

Realize testes e coloque nos comentários quais foram os resultados, suas observações são valiósas.

Até o próximo artigo!

DEV Community: Kafka BR

Cotas de Cliente no Apache Kafka® 2.6.0

Artigo publicado no Blog oficial Kafka BR

Transações no Apache Kafka®

Artigo publicado no Blog oficial Kafka BR.

Kafka: produtores idempotentes

Apagar eventos do kafka?

Como implementar Dead-letter Topic com Spring Kafka

fabiojose / spring-kafka-dlt-ex

Dead-letter Topic com Spring Kafka

Classificar Erros

Recuperáveis

Não-recuperáveis

Implementação

Tópicos

Construção

fabiojose / spring-kafka-dlt-ex

Dead-letter Topic com Spring Kafka

spring-kafka-dlt-ex

Requerimentos

Configurações

Variáveis de Ambiente

Tratamento

Kafka: consumindo registros com Spring

Resolvendo

fabiojose / skc-ex

Spring Kafka Consumer

Spring Kafka Consumer Example

Requerimentos

Build & Run

Maven

Docker

Kafka: compactação e compressão

Compactação

Compressão

Kafka: Garantia de Ordem

Como é a ordenação no Kafka

O que pode mudar a ordenação

Configuração correta

Monitorando Consumer Lag do Apache Kafka

Instalando e Configurando o Prometheus

wget https://github.com/prometheus/prometheus/releases/download/v2.12.0/prometheus-2.12.0.linux-amd64.tar.gz

tar -zxvf prometheus-2.12.0.linux-amd64.tar.gz -C /srv/

useradd -s /usr/sbin/nologin prometheus

chown prometheus:prometheus /srv/prometheus-2.12.0.linux-amd64/ -R

vim /etc/systemd/system/prometheus.service

systemctl daemon-reload

systemctl start prometheus

systemctl start grafana-server

alvarobacelar / burrow_exporter

A Prometheus Exporter for gathering Kafka consumer group info from Burrow

Conclusão

Kafka: quantas partições por tópico?

Aplicando a fórmula