Enviar y recibir mensajes Kafka en Spring Boot

1. Descripción general

Apache Kafka es una plataforma de event streaming popular para que los servicios se comuniquen de forma asíncrona. Spring Boot soporta Kafka a través de la biblioteca Spring Kafka, lo que te evita tener que gestionar manualmente KafkaProducer y el bucle poll().

En este artículo, crearás un ejemplo pequeño: un servicio de pedidos envía un evento OrderCreatedEvent, y un consumer recibe ese evento para reservar stock en el almacén. Después de leerlo, sabrás cómo configurar, enviar y recibir mensajes en formato JSON, manejar errores con Dead Letter Topic y escribir pruebas con Embedded Kafka.

2. Preparación

2.1. Dependencia Maven

Necesitas spring-kafka. Spring Boot ya gestiona la versión, así que no es necesario especificarla:

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

<!-- Para pruebas -->
<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 proporciona un broker Kafka que se ejecuta en memoria para las pruebas. Awaitility ayuda a esperar el procesamiento asíncrono sin tener que usar Thread.sleep().

2.2. Ejecutar Kafka localmente

La forma más rápida es usar la imagen oficial de Apache Kafka en modo KRaft, es decir, sin necesidad de ZooKeeper:

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

El broker escuchará en localhost:9092.

3. Modelo de datos y configuración

El evento que se envía es un record simple:

package com.nonla.shop.order;

import java.math.BigDecimal;

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

A continuación está la configuración en 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"

Significado de cada parte:

• bootstrap-servers es la dirección del broker.
• El productor usa StringSerializer para la clave y JsonSerializer para el valor. JsonSerializer usa Jackson para convertir objetos a JSON.
• acks: all obliga al broker a reconocer solo después de que todas las réplicas sincronizadas han escrito. Esta forma es más segura pero un poco más lenta.
• group-id define el grupo de consumidores. Las instancias del mismo grupo compartirán las particiones.
• auto-offset-reset: earliest permite al consumidor leer desde el principio del tema cuando el grupo aún no ha confirmado ningún offset.
• spring.json.trusted.packages declara los paquetes en los que JsonDeserializer puede crear objetos. Si falta esta línea, obtendrás el error “The class … is not in the trusted packages”.

3.1. Crear temas automáticamente

Spring Boot configura automáticamente un KafkaAdmin. Este bean creará todos los NewTopic declarados en el 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();
    }
}

Cuando la aplicación se inicia, el tema orders con 3 particiones se creará si aún no existe. Si el tema ya existe, KafkaAdmin lo ignorará y no sobrescribirá la configuración actual.

4. Enviar mensajes con KafkaTemplate

Spring Boot crea automáticamente el bean KafkaTemplate basado en la configuración del productor. Solo necesitas inyectarlo:

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("Falló al enviar el pedido {}", event.orderId(), ex);
                        return;
                    }
                    var metadata = result.getRecordMetadata();
                    log.info("Se envió el pedido {} a la partición {} offset {}",
                            event.orderId(), metadata.partition(), metadata.offset());
                });
    }
}

Puntos a tener en cuenta:

• send(topic, key, value) usa orderId como clave. Kafka hash la clave para elegir la partición, así que todos los mensajes de un mismo pedido irán a la misma partición. Gracias a esto, se mantiene el orden entre ellos.
• Desde Spring Kafka 3.0, send() devuelve CompletableFuture<SendResult> en lugar de ListenableFuture, que ha sido eliminado.
• send() es asíncrono. La llamada retorna de inmediato, mientras que el callback en whenComplete se ejecuta cuando el broker responde.

Cuando se envía con éxito, el log tendrá este aspecto:

INFO  c.n.s.order.OrderEventPublisher : Se envió el pedido ORD-1001 a la partición 2 offset 0

Si necesitas bloquear el hilo hasta que se complete el envío, por ejemplo en un trabajo por lotes, puedes llamar a publish(event).get(5, TimeUnit.SECONDS). No hagas esto en hilos de procesamiento de solicitudes con carga alta, porque cada llamada mantendrá un hilo esperando al broker.

Nota: Escribir en la base de datos y luego enviar a Kafka en el mismo método no es una transacción atómica. Si la confirmación de la base de datos tiene éxito pero el envío a Kafka falla, los datos se desincronizarán. Para datos críticos, debes considerar el patrón Transactional Outbox.

5. Recibir mensajes con @KafkaListener

El lado del consumidor solo necesita un método con la anotación @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("Se recibió el pedido {} de la partición {} offset {}", event.orderId(), partition, offset);
        inventoryService.reserve(event);
    }
}

InventoryService es solo un servicio ordinario:

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("Monto no válido: " + event.amount());
        }
        // Reservar stock en el almacén...
    }
}

Explicación:

• topics especifica el tema a escuchar. groupId se toma de spring.kafka.consumer.group-id, o puedes anularlo directamente en la anotación.
• concurrency = "3" crea 3 hilos de consumidor, cada hilo maneja una partición. Un valor mayor que el número de particiones es inútil, porque los hilos sobrantes no recibirán nada.
• @Payload recibe el objeto deserializado. @Header obtiene metadatos como partición y offset.
• Por defecto, el offset se confirma después de que el método termina sin lanzar excepciones (AckMode.BATCH). Por lo tanto, si la aplicación se detiene inesperadamente, el mensaje se procesará nuevamente. Esta es la semántica at-least-once, por lo que tu lógica debe ser idempotente: procesar el mismo orderId dos veces no debe reservar inventario dos veces.

Si necesitas la clave, timestamp o encabezados personalizados, puedes recibir un ConsumerRecord<String, OrderCreatedEvent> completo como parámetro.

6. Manejo de errores y Dead Letter Topic

Cuando el listener lanza una excepción, DefaultErrorHandler por defecto reintentará 9 veces seguidas, registrará en el log y luego ignorará el mensaje. En la práctica, generalmente quieres enviar mensajes con errores a un tema separado para inspeccionar más tarde:

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;
    }
}

Spring Boot automáticamente vincula este bean CommonErrorHandler a la fábrica de contenedores por defecto. La configuración anterior funciona de la siguiente manera:

• FixedBackOff(1000L, 2) reintenta como máximo 2 veces, con un intervalo de 1 segundo entre intentos.
• Cuando se agotan los intentos, DeadLetterPublishingRecoverer envía el mensaje al tema orders.DLT. Este nombre es el nombre original más el sufijo .DLT, y el mensaje mantiene su partición. El tema DLT por lo tanto debe tener al menos el mismo número de particiones que el tema original.
• IllegalArgumentException es un error de datos, así que reintentar no ayudaría. El controlador enviará directamente el mensaje a DLT.

Un mensaje con amount = -5 producirá un log similar a:

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

Nota: Los errores que ocurren durante la deserialización, por ejemplo JSON malformado, no llegarán al listener. Si no se manejan, el consumidor puede quedarse atrapado en ese mensaje indefinidamente, también llamado “poison pill”. Envuelve el deserializador con ErrorHandlingDeserializer:

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

Con esta configuración, los errores de deserialización se envían a DefaultErrorHandler y el mensaje de error también va a DLT.

7. Pruebas con Embedded Kafka

@EmbeddedKafka inicia un broker en memoria, así que no necesitas Docker al ejecutar pruebas. El ejemplo a continuación verifica el flujo completo: envía a través de OrderEventPublisher y luego verifica que InventoryService haya recibido el evento correcto.

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);
    }
}

Qué hace cada parte:

• ${spring.embedded.kafka.brokers} es una propiedad proporcionada por @EmbeddedKafka. Contiene la dirección del broker embebido, y la usamos para anular bootstrap-servers.
• @MockitoBean (disponible desde Spring Boot 3.4) reemplaza el InventoryService real con un mock para facilitar la verificación. Con versiones anteriores, usas @MockBean.
• publish(event).get() espera la confirmación del broker, así que la prueba se asegura de que el mensaje se haya escrito.
• El listener se ejecuta en un hilo diferente, por lo que await().untilAsserted(...) reintentará la verificación durante un máximo de 10 segundos.
• Porque OrderCreatedEvent es un record, equals() compara por valor. assertThat(...).isEqualTo(event) por lo tanto verifica que el JSON se haya serializado y deserializado correctamente.

Es auto-offset-reset: earliest en la configuración lo que ayuda a que las pruebas sean estables. Si el consumidor logra unirse al grupo después de que el mensaje se haya enviado, aún puede leer ese mensaje.

8. Errores comunes y cuándo no usar Kafka

Algunos errores que es probable que encuentres:

• Falta trusted.packages: JsonDeserializer se niega a crear clases desconocidas. Declara correctamente el paquete. Solo usa "*" cuando controlas completamente todos los productores.
• Productor y consumidor en paquetes diferentes: JsonSerializer por defecto escribe el nombre de clase completo en el encabezado __TypeId__. Si el servicio del consumidor tiene una clase con el mismo nombre pero en un paquete diferente, la deserialización fallará. En ese caso, establece spring.json.use.type.headers: false y spring.json.value.default.type: com.nonla.inventory.OrderCreatedEvent en el lado del consumidor.
• El listener tarda mucho: si un poll() tarda más que max.poll.interval.ms (5 minutos por defecto), el broker considera que el consumidor está muerto y rebalancea. Como resultado, los mensajes se procesan duplicados. Reduce max.poll.records u optimiza la lógica.
• concurrency mayor que el número de particiones: los hilos sobrantes estarán inactivos. Si quieres aumentar el rendimiento, primero debes aumentar el número de particiones.
• Enviar null como clave: el mensaje se distribuye uniformemente entre las particiones, por lo que pierdes la garantía de orden por entidad.

Kafka tampoco es la mejor opción en todos los casos:

• Necesitas solicitud/respuesta sincrónica y resultados inmediatos. En ese caso, usa REST o gRPC.
• Sistema pequeño, bajo volumen y no necesitas histórico de eventos. Operar un cluster Kafka en ese caso es un gasto innecesario, y RabbitMQ o una tabla de cola en la base de datos puede ser suficiente.
• Necesitas enrutamiento complejo por mensaje, o latencia o prioridad específica por mensaje. Esta es la fortaleza de un message broker tradicional.

9. Conclusión

En este artículo, configuraste Spring Kafka con un serializador JSON y creaste un tema usando NewTopic. Enviaste mensajes a través de KafkaTemplate con una clave para mantener el orden, y recibiste mensajes con @KafkaListener con concurrency. También manejaste errores con DefaultErrorHandler combinado con Dead Letter Topic, y escribiste una prueba de integración con @EmbeddedKafka y Awaitility.

Cuando lo pongas en producción, recuerda estos tres puntos: el listener debe ser idempotente, debes manejar errores de deserialización con ErrorHandlingDeserializer, y el envío de eventos debe ser consistente con los datos en la base de datos.