DEV Community: Fabio José

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.

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!