Enviando e recebendo mensagens Kafka no Spring Boot

1. Visão geral

Apache Kafka é uma plataforma de event streaming popular para serviços se comunicarem de forma assíncrona. O Spring Boot suporta Kafka através da biblioteca Spring Kafka, ajudando você a não precisar gerenciar manualmente KafkaProducer e loops poll().

Neste artigo, você construirá um pequeno exemplo: um serviço de pedidos envia um evento OrderCreatedEvent, e um consumidor recebe esse evento para reservar produtos no inventário. Após ler, você saberá como configurar, enviar e receber mensagens em formato JSON, tratar erros com Dead Letter Topic e testar com Embedded Kafka.

2. Preparação

2.1. Dependência Maven

Você precisa de spring-kafka. O Spring Boot já gerencia a versão, então você não precisa especificá-la:

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

<!-- Para testes -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
    <scope>test</scope>
</dependency>

spring-kafka-test fornece um broker Kafka em memória para testes. Awaitility ajuda a aguardar processamento assíncrono sem precisar usar Thread.sleep().

2.2. Executar Kafka localmente

A forma mais rápida é usar a imagem oficial do Apache Kafka em modo KRaft, ou seja, sem precisar de ZooKeeper:

docker run -d --name kafka -p 9092:9092 apache/kafka:latest

O broker será ouvido em localhost:9092.

3. Modelo de dados e configuração

O evento a ser enviado é um record simples:

package com.nonla.shop.order;

import java.math.BigDecimal;

public record OrderCreatedEvent(String orderId, String customerId, BigDecimal amount) {
}

Em seguida, vem a configuração em application.yml:

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all
    consumer:
      group-id: inventory-service
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "com.nonla.shop.order"

O significado de cada parte:

• bootstrap-servers é o endereço do broker.
• O produtor usa StringSerializer para a chave e JsonSerializer para o valor. JsonSerializer usa Jackson para converter objetos em JSON.
• acks: all força o broker a confirmar apenas depois que todas as réplicas sincronizadas tiverem escrito. Isso é mais seguro, mas um pouco mais lento.
• group-id define o grupo de consumidores. As instâncias do mesmo grupo compartilham as partições.
• auto-offset-reset: earliest faz o consumidor ler desde o início do tópico quando o grupo não tem nenhum offset confirmado.
• spring.json.trusted.packages declara os pacotes nos quais JsonDeserializer pode criar objetos. Se você omitir isso, terá um erro “The class … is not in the trusted packages”.

3.1. Criar tópicos automaticamente

O Spring Boot configura automaticamente um KafkaAdmin. Este bean criará cada NewTopic declarado no contexto:

package com.nonla.shop.config;

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;

@Configuration
public class KafkaTopicConfig {

    public static final String ORDERS_TOPIC = "orders";

    @Bean
    public NewTopic ordersTopic() {
        return TopicBuilder.name(ORDERS_TOPIC)
                .partitions(3)
                .replicas(1)
                .build();
    }
}

Quando a aplicação inicia, o tópico orders com 3 partições será criado se não existir. Se o tópico já existe, KafkaAdmin o ignora e não sobrescreve a configuração atual.

4. Enviando mensagens com KafkaTemplate

O Spring Boot cria automaticamente o bean KafkaTemplate com base na configuração do produtor. Você apenas precisa injetá-lo:

package com.nonla.shop.order;

import com.nonla.shop.config.KafkaTopicConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Service
public class OrderEventPublisher {

    private static final Logger log = LoggerFactory.getLogger(OrderEventPublisher.class);

    private final KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate;

    public OrderEventPublisher(KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public CompletableFuture<SendResult<String, OrderCreatedEvent>> publish(OrderCreatedEvent event) {
        return kafkaTemplate.send(KafkaTopicConfig.ORDERS_TOPIC, event.orderId(), event)
                .whenComplete((result, ex) -> {
                    if (ex != null) {
                        log.error("Falha ao enviar pedido {}", event.orderId(), ex);
                        return;
                    }
                    var metadata = result.getRecordMetadata();
                    log.info("Enviado pedido {} para partição {} offset {}",
                            event.orderId(), metadata.partition(), metadata.offset());
                });
    }
}

Pontos a notar:

• send(topic, key, value) usa orderId como chave. Kafka faz hash da chave para escolher a partição, então todas as mensagens do mesmo pedido vão para a mesma partição. Assim, a ordem entre elas é preservada.
• A partir do Spring Kafka 3.0, send() retorna CompletableFuture<SendResult> em vez de ListenableFuture que foi descontinuado.
• send() é assíncrono. A chamada retorna imediatamente, enquanto o callback em whenComplete executa quando o broker responde.

Quando bem-sucedido, o log terá uma aparência assim:

INFO  c.n.s.order.OrderEventPublisher : Enviado pedido ORD-1001 para partição 2 offset 0

Se você precisar bloquear a thread até a conclusão do envio, por exemplo em um trabalho em lote, pode chamar publish(event).get(5, TimeUnit.SECONDS). Não faça isso no processamento de requisições com alta carga, pois cada chamada manterá uma thread aguardando o broker.

Nota: Escrever no banco de dados e enviar para Kafka no mesmo método não é uma transação atômica. Se o commit do banco de dados for bem-sucedido mas o envio para Kafka falhar, os dados ficarão inconsistentes. Para dados críticos, considere o padrão Transactional Outbox.

5. Recebendo mensagens com @KafkaListener

No lado do consumidor, você precisa apenas de um método com a anotação @KafkaListener:

package com.nonla.shop.inventory;

import com.nonla.shop.config.KafkaTopicConfig;
import com.nonla.shop.order.OrderCreatedEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

@Component
public class OrderCreatedListener {

    private static final Logger log = LoggerFactory.getLogger(OrderCreatedListener.class);

    private final InventoryService inventoryService;

    public OrderCreatedListener(InventoryService inventoryService) {
        this.inventoryService = inventoryService;
    }

    @KafkaListener(topics = KafkaTopicConfig.ORDERS_TOPIC, concurrency = "3")
    public void onOrderCreated(@Payload OrderCreatedEvent event,
                               @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
                               @Header(KafkaHeaders.OFFSET) long offset) {
        log.info("Recebido pedido {} de partição {} offset {}", event.orderId(), partition, offset);
        inventoryService.reserve(event);
    }
}

InventoryService é apenas um serviço comum:

package com.nonla.shop.inventory;

import com.nonla.shop.order.OrderCreatedEvent;
import org.springframework.stereotype.Service;

@Service
public class InventoryService {

    public void reserve(OrderCreatedEvent event) {
        if (event.amount().signum() <= 0) {
            throw new IllegalArgumentException("Valor inválido: " + event.amount());
        }
        // Reservar produto no inventário...
    }
}

Explicação:

• topics especifica o tópico a ouvir. groupId vem de spring.kafka.consumer.group-id, ou você pode sobrescrever diretamente na anotação.
• concurrency = "3" cria 3 threads de consumidor, cada um responsável por uma partição. Valores maiores que o número de partições são inúteis, pois threads extras não receberão nada.
• @Payload recebe o objeto já desserializado. @Header obtém metadados como partição e offset.
• Por padrão, o offset é confirmado após o método ser executado sem lançar uma exceção (AckMode.BATCH). Portanto, se a aplicação morrer no meio do caminho, a mensagem será processada novamente. Esta é a semântica at-least-once, então sua lógica deve ser idempotente: processar o mesmo orderId duas vezes não deve reservar produto duas vezes.

Se você precisar da chave, timestamp ou headers customizados, pode receber o ConsumerRecord<String, OrderCreatedEvent> completo como parâmetro.

6. Tratando erros e Dead Letter Topic

Quando o listener lança uma exceção, DefaultErrorHandler padrão retentará 9 vezes seguidas, depois registrará e ignorará a mensagem. Na prática, você geralmente quer enviar mensagens de erro para um tópico separado para inspecionar depois:

package com.nonla.shop.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.util.backoff.FixedBackOff;

@Configuration
public class KafkaErrorConfig {

    @Bean
    public DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> kafkaTemplate) {
        var recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate);
        var handler = new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 2));
        handler.addNotRetryableExceptions(IllegalArgumentException.class);
        return handler;
    }
}

O Spring Boot vincula automaticamente este bean CommonErrorHandler à fábrica de contêineres padrão. A configuração acima funciona da seguinte forma:

• FixedBackOff(1000L, 2) retenta no máximo 2 vezes, com intervalo de 1 segundo entre cada uma.
• Quando as tentativas se esgotam, DeadLetterPublishingRecoverer envia a mensagem para o tópico orders.DLT. Este nome é o nome original mais o sufixo .DLT, e a mensagem mantém a partição. O tópico DLT portanto precisa ter pelo menos tantas partições quanto o tópico original.
• IllegalArgumentException é um erro de dados, então retry é inútil. O handler enviará a mensagem diretamente para DLT.

Uma mensagem com amount = -5 produzirá um log similar:

ERROR o.s.k.l.DefaultErrorHandler : Backoff none exhausted for orders-1@4
INFO  c.n.s.inventory.DltListener  : Mensagem de erro em DLT: ORD-1002

Nota: Erros que ocorrem durante desserialização, por exemplo JSON malformado, não chegam ao listener. Se não tratado, o consumidor pode ficar preso nessa mensagem para sempre, também chamado de “poison pill”. Envolva o desserializador com ErrorHandlingDeserializer:

spring:
  kafka:
    consumer:
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer

Com essa configuração, erros de desserialização são passados para DefaultErrorHandler e a mensagem de erro também vai para DLT.

7. Testando com Embedded Kafka

@EmbeddedKafka inicia um broker em memória, então você não precisa do Docker quando executa testes. O exemplo abaixo verifica o fluxo completo: enviar via OrderEventPublisher e confirmar que InventoryService recebeu o evento corretamente.

package com.nonla.shop.order;

import com.nonla.shop.inventory.InventoryService;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.test.context.bean.override.mockito.MockitoBean;

import java.math.BigDecimal;
import java.time.Duration;

import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import static org.mockito.Mockito.verify;

@SpringBootTest(properties = "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}")
@EmbeddedKafka(partitions = 3, topics = "orders")
class OrderKafkaIntegrationTest {

    @Autowired
    private OrderEventPublisher publisher;

    @MockitoBean
    private InventoryService inventoryService;

    @Test
    void whenOrderPublished_thenListenerReservesInventory() throws Exception {
        var event = new OrderCreatedEvent("ORD-1001", "CUS-42", new BigDecimal("250000"));

        var result = publisher.publish(event).get();
        assertThat(result.getRecordMetadata().topic()).isEqualTo("orders");

        var captor = ArgumentCaptor.forClass(OrderCreatedEvent.class);
        await().atMost(Duration.ofSeconds(10))
                .untilAsserted(() -> verify(inventoryService).reserve(captor.capture()));

        assertThat(captor.getValue()).isEqualTo(event);
    }
}

O que cada parte faz:

• ${spring.embedded.kafka.brokers} é uma propriedade fornecida por @EmbeddedKafka. Ela contém o endereço do broker incorporado, e a usamos para sobrescrever bootstrap-servers.
• @MockitoBean (disponível a partir do Spring Boot 3.4) substitui o InventoryService real por um mock para facilitar a verificação. Em versões anteriores, use @MockBean.
• publish(event).get() aguarda a confirmação do broker, garantindo que a mensagem foi escrita.
• O listener é executado em outra thread, então await().untilAsserted(...) retentará a verificação por até 10 segundos.
• Como OrderCreatedEvent é um record, equals() compara por valor. assertThat(...).isEqualTo(event) portanto confirma que o JSON foi serializado e desserializado corretamente.

É o auto-offset-reset: earliest na configuração que torna o teste estável. Se o consumidor se juntar ao grupo depois que a mensagem foi enviada, ele ainda consegue ler a mensagem.

8. Erros comuns e quando não usar Kafka

Alguns erros que você pode encontrar:

• Ausência de trusted.packages: JsonDeserializer se recusa a criar classes desconhecidas. Declare o pacote correto. Apenas use "*" quando você controlar completamente todos os produtores.
• Produtor e consumidor em pacotes diferentes: JsonSerializer por padrão escreve o nome completo da classe no header __TypeId__. Se o serviço consumidor tem uma classe com o mesmo nome mas em um pacote diferente, a desserialização falhará. Neste caso, defina spring.json.use.type.headers: false e spring.json.value.default.type: com.nonla.inventory.OrderCreatedEvent no lado do consumidor.
• Listener levando muito tempo: se um único poll() levar mais que max.poll.interval.ms (padrão 5 minutos), o broker considerará o consumidor morto e fará rebalanceamento. O resultado é o processamento duplicado de mensagens. Reduza max.poll.records ou otimize a lógica.
• concurrency maior que o número de partições: threads extras ficarão ociosas. Para aumentar a taxa de transferência, você deve aumentar o número de partições primeiro.
• Enviar chave null: a mensagem é distribuída uniformemente entre as partições, então você perde a garantia de ordem por entidade.

Kafka também não é a melhor escolha em todos os casos:

• Você precisa de request/response síncrono com resultados imediatos. Neste caso, use REST ou gRPC.
• Sistema pequeno, baixo volume e sem necessidade de reter histórico de eventos. Operar um cluster Kafka seria um custo excessivo, e RabbitMQ ou uma tabela de fila no banco de dados pode ser suficiente.
• Você precisa de roteamento complexo por mensagem, ou de latência ou prioridade específica por mensagem. Estes são os pontos fortes dos message brokers tradicionais.

9. Conclusão

Neste artigo, você configurou o Spring Kafka com serialização JSON e criou tópicos com NewTopic. Você enviou mensagens via KafkaTemplate com chaves para manter a ordem, recebeu mensagens com @KafkaListener com concurrency. Você também tratou erros com DefaultErrorHandler combinado com Dead Letter Topic, e escreveu testes de integração com @EmbeddedKafka e Awaitility.

Ao colocar em produção, lembre-se de três pontos: o listener deve ser idempotente, deve lidar com erros de desserialização com ErrorHandlingDeserializer, e o envio de eventos deve ser consistente com os dados no banco de dados.