Spring Boot y Kafka – Ejemplos Prácticos

Este post de blog muestra cómo configurar Spring Kafka y Spring Boot para enviar mensajes usando JSON y recibirlos en múltiples formatos: JSON, Strings planos, o arrays de bytes.

Esta aplicación de ejemplo demuestra cómo usar múltiples consumidores de Kafka dentro del mismo consumer group con la anotación @KafkaListener para que los mensajes se balanceen de carga. Cada consumidor implementa un enfoque de deserialización diferente.

Además, este post incluye ejercicios prácticos ajustando Consumer Group y particiones de Topic para ayudarte a comprender los conceptos de Kafka.

Actualizado (8 de Mayo, 2025): Este post usa Spring Boot 3.4 y Java 24. Si tienes curiosidad sobre los cambios, revisa el Pull Request.

Múltiples consumidores en un consumer group

Vista Lógica

Para entender mejor la configuración que usaremos en los ejemplos, echa un vistazo al diagrama debajo. Como puedes ver, creamos un topic de Kafka con tres particiones. Solo hay una aplicación en el lado del consumidor, pero implementa tres consumidores de Kafka con la misma propiedad group.id. Esta configuración es necesaria para tenerlos en el mismo Kafka Consumer Group.

Cuando iniciamos la aplicación, Kafka asigna a cada consumidor una partición diferente. Este consumer group recibirá los mensajes de manera balanceada, actuando efectivamente como un único consumidor conceptual con tres workers paralelos. Más adelante en este post, verás la diferencia si hacemos que tengan identificadores de grupo diferentes (probablemente conoces el resultado si estás familiarizado con Kafka).

El Caso de Uso de Ejemplo

La lógica que vamos a construir es simple. Cada vez que llamemos a un endpoint REST dado, /hello, la app producirá un número configurable de mensajes y los enviará al mismo topic, usando un número de secuencia como la clave de Kafka. La aplicación esperará (usando un CountDownLatch) a que todos los mensajes sean consumidos antes de retornar un mensaje, Hello Kafka!. Habrá tres consumidores, cada uno usando un mecanismo de deserialización diferente, que disminuirán el contador del latch cuando reciban un nuevo mensaje.

Fácil, ¿verdad? Veamos cómo construirlo.

Configurando Spring Boot y Kafka

Todo el código en este post está disponible en GitHub: Kafka and Spring Boot Example. Si lo encuentras útil, ¡por favor dale una estrella!

Iniciando Kafka

Primero, necesitas tener un cluster de Kafka ejecutándose para conectarte. Usaré Docker Compose y Kafka ejecutándose en un contenedor de un solo nodo para esta aplicación. Esto está lejos de ser una configuración de producción, pero es lo suficientemente bueno para el objetivo de este post.

docker-compose.yml

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.9.0
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "2181:2181"
    healthcheck:
      test: ["CMD", "nc", "-z", "localhost", "2181"]
      interval: 10s
      timeout: 5s
      retries: 5
  kafka:
    image: confluentinc/cp-kafka:7.9.0
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENERS: LISTENER_DOCKER://0.0.0.0:29092,LISTENER_HOST://0.0.0.0:9092
      KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER://kafka:29092,LISTENER_HOST://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER:PLAINTEXT,LISTENER_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'false'
    healthcheck:
      test: ["CMD", "nc", "-z", "localhost", "9092"]
      interval: 10s
      timeout: 5s
      retries: 5

  kafka-ui:
    image: provectuslabs/kafka-ui:v0.7.2
    container_name: kafka-ui
    depends_on:
      - kafka
    ports:
      - "9080:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: docker_local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:29092
      KAFKA_CLUSTERS_0_ZOOKEEPER: zookeeper:2181

Esta configuración instruye a Docker Compose a iniciar tres contenedores: el nodo broker de Kafka, Zookeeper (requerido para Kafka), y uno opcional exponiendo una interfaz de usuario para monitorear y gestionar Kafka (kafka-ui).

El nodo de Kafka se conecta a Zookeeper dentro de la red de Docker y expone un listener a la red externa (el computador host). Si quieres aprender más sobre esta configuración básica de listeners de Kafka, revisa este artículo de Confluent.

El contenedor Kafka UI está disponible en localhost:9080 y te da una buena vista general de nuestra configuración, topics, mensajes, etc. Incluso tiene alguna funcionalidad para resetear los offsets, enviar mensajes, etc. Te recomiendo echar un vistazo a esta herramienta y navegar a través de la configuración que crearemos.

No profundicemos en más detalles. Por ahora, solo nota que configuré Kafka para no crear topics automáticamente (KAFKA_AUTO_CREATE_TOPICS_ENABLE: ‘false’). Crearemos nuestro topic desde la aplicación Spring Boot ya que queremos especificar alguna configuración personalizada.

Para iniciar nuestros contenedores, solo ejecuta docker-compose up desde la carpeta donde vive este archivo.

Aplicación básica de Spring Boot y Kafka

La forma más fácil de obtener un esqueleto para nuestra app es navegar a start.spring.io, llenar los detalles básicos para nuestro proyecto, y seleccionar Kafka como dependencia. Luego, descarga el archivo zip y usa tu IDE favorito para cargar las fuentes.

Usemos YAML para nuestra configuración. Puede que necesites renombrar el archivo application.properties dentro de src/main/java/resources a application.yml. Estos son los valores de configuración que vamos a usar para esta aplicación de ejemplo:

application.yml

spring:
  kafka:
    consumer:
      group-id: tpd-loggers
      auto-offset-reset: earliest
    # change this property if you are using your own
    # Kafka cluster or your Docker IP is different
    bootstrap-servers: localhost:9092

tpd:
  topic-name: advice-topic
  messages-per-request: 10

El primer bloque de propiedades es la configuración de Spring Kafka:

• El group-id que nuestros consumidores usan por defecto.
• La propiedad auto-offset-reset está establecida en earliest. Esto significa que los consumidores comenzarán a leer mensajes desde el más temprano disponible cuando no exista un offset existente para ese consumidor.
• En este caso, el servidor de Kafka es el único disponible ya que usamos la configuración de un solo nodo. Nota que esta propiedad es redundante si usas el valor por defecto, localhost:9092. La agregué explícitamente allí para aclararlo y hacerte saber que debes cambiarla si tu Docker está mapeado a una IP diferente.

El segundo bloque es específico de la aplicación. Definimos el nombre del topic de Kafka y el número de mensajes a enviar cuando se hace una llamada a la API REST HTTP.

La clase Message

Este es el record de Java que usaremos como mensaje de Kafka. Necesitamos usar las anotaciones @JsonProperty para los campos del record para que Jackson pueda deserializar el mensaje correctamente. Si estás usando una versión de Java que no soporta records, puedes usar la versión antigua de este archivo disponible en GitHub.

PracticalAdvice.java

package io.tpd.kafkaexample;

import com.fasterxml.jackson.annotation.JsonProperty;

record PracticalAdvice(@JsonProperty("message") String message,
                       @JsonProperty("identifier") int identifier) {
}

Configuración del Producer de Kafka en Spring Boot

Agregaremos la configuración en la clase principal de Spring Boot para mantener la aplicación simple. Eventualmente, queremos incluir tanto las configuraciones de producer como de consumer aquí y usar tres variaciones diferentes para deserialización. Recuerda que puedes encontrar el código fuente completo en el repositorio de GitHub.

Primero, enfoquémonos en la configuración del Producer.

KafkaExampleApplication.java

@SpringBootApplication
public class KafkaExampleApplication {

    public static void main(String[] args) {
        SpringApplication.run(KafkaExampleApplication.class, args);
    }

    @Autowired
    private KafkaProperties kafkaProperties;

    @Value("${tpd.topic-name}")
    private String topicName;

    // Producer configuration

    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props =
                new HashMap<>(kafkaProperties.buildProducerProperties());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                JsonSerializer.class);
        return props;
    }

    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    @Bean
    public NewTopic adviceTopic() {
        return new NewTopic(topicName, 3, (short) 1);
    }
}

En esta configuración, estamos configurando dos partes de la aplicación:

• La instancia KafkaTemplate. Este es el objeto que empleamos para enviar mensajes a Kafka. No queremos usar la predeterminada porque queremos jugar con diferentes configuraciones, así que necesitamos inyectar nuestra versión personalizada en el contexto de aplicación de Spring.

  • Tipamos (con genéricos) el KafkaTemplate para tener una clave String plana y un Object como valor. La razón para tener Object como valor es que queremos enviar múltiples tipos de objetos con la misma plantilla.
  • El KafkaTemplate acepta un ProducerFactory como parámetro. Creamos una factory por defecto con nuestros valores de configuración personalizados del producer.
  • La Configuración del Producer es un mapa simple de clave-valor. Inyectamos las propiedades por defecto usando @Autowired para obtener el bean KafkaProperties. Luego, construimos nuestro mapa, pasando los valores por defecto para el producer y sobrescribiendo los serializadores por defecto de clave y valor de Kafka. El producer serializará las claves como Strings usando el StringSerializer de la librería de Kafka y hará lo mismo para los valores, pero esta vez usando JSON, con un JsonSerializer (proporcionado por Spring Kafka).

• El topic de Kafka. Cuando inyectamos un bean NewTopic, estamos instruyendo al bean AdminClient de Kafka (ya en el contexto) a crear un topic con la configuración dada. El primer parámetro es el nombre (advice-topic), que usamos en la configuración de la app. El segundo parámetro es el número de particiones (3) y el tercero es el factor de replicación (1, ya que estamos usando un solo nodo de todos modos).

Sobre Serializadores y Deserializadores de Kafka para Java

La librería core de Kafka (Javadoc) proporciona algunos serializadores básicos para Strings, todas las clases numéricas, y arrays de bytes, así como los de JSON proporcionados por Spring Kafka (Javadoc).

Además de eso, puedes crear tus Serializadores y Deserializadores solo implementando Serializer o ExtendedSerializer, o sus versiones correspondientes para deserialización. Eso te da mucha flexibilidad para optimizar la cantidad de datos viajando a través de Kafka si necesitas hacerlo. Como puedes ver en esas interfaces, Kafka trabaja con arrays de bytes planos, así que, eventualmente, no importa con qué tipo complejo estés trabajando, necesita ser transformado en un byte[].

Con ese entendimiento, puede que te preguntes por qué alguien querría usar JSON con Kafka. Es ineficiente ya que estás transformando tus objetos a JSON y luego a un array de bytes. Pero tienes que considerar dos ventajas principales de hacer esto:

• Un humano puede leer JSON más fácilmente que un array de bytes. Si quieres depurar o analizar los contenidos de tus topics de Kafka, será mucho más directo que mirar bytes desnudos.
• JSON es un estándar, mientras que los serializadores de array de bytes por defecto dependen de la implementación del lenguaje de programación. Así, para consumir mensajes desde múltiples lenguajes de programación, debes replicar la lógica del (de)serializador en todos esos lenguajes.

Por otro lado, si te preocupa la carga de tráfico en Kafka, el almacenamiento, o la velocidad en (de)serialización, puede que quieras elegir arrays de bytes o incluso implementar tu propio serializador/deserializador.

Enviando mensajes con Spring Boot y Kafka

Siguiendo el plan, creamos un Rest Controller y usamos el KafkaTemplate inyectado para producir algunos mensajes JSON cuando un cliente llama al endpoint.

Esta es la primera implementación del controller, conteniendo solo la lógica que produce los mensajes.

HelloKafkaController.java

@RestController
public class HelloKafkaController {

    private static final Logger logger =
            LoggerFactory.getLogger(HelloKafkaController.class);

    private final KafkaTemplate<String, Object> template;
    private final String topicName;
    private final int messagesPerRequest;
    private CountDownLatch latch;

    public HelloKafkaController(
            final KafkaTemplate<String, Object> template,
            @Value("${tpd.topic-name}") final String topicName,
            @Value("${tpd.messages-per-request}") final int messagesPerRequest) {
        this.template = template;
        this.topicName = topicName;
        this.messagesPerRequest = messagesPerRequest;
    }

    @GetMapping("/hello")
    public String hello() throws Exception {
        latch = new CountDownLatch(messagesPerRequest);
        IntStream.range(0, messagesPerRequest)
                .forEach(i -> this.template.send(topicName, String.valueOf(i),
                        new PracticalAdvice("A Practical Advice", i))
                );
        latch.await(60, TimeUnit.SECONDS);
        logger.info("All messages received");
        return "Hello Kafka!";
    }
}

En el constructor, pasamos algunos parámetros de configuración y el KafkaTemplate que personalizamos para enviar claves String y valores JSON. Luego, cuando el cliente de API solicita el endpoint /hello, enviamos 10 mensajes (ese es el valor de configuración de messagesPerRequest), y luego bloqueamos el hilo por un máximo de 60 segundos. Como puedes ver, aún no hay implementación para que los consumidores de Kafka disminuyan el contador del latch. Una vez que el latch se desbloquea, retornamos el mensaje Hello Kafka! a nuestro cliente.

Esta idea de usar un latch no es un patrón de la vida real, pero es adecuado para que aprendamos cómo funcionan los producers y consumers. De esa manera, puedes verificar el número de mensajes recibidos. Si prefieres, puedes remover el latch y retornar el mensaje “Hello Kafka!” antes de recibir los mensajes. Ese sería un caso más realista donde el flujo sucede asincrónicamente.

Configuración del Consumer de Kafka

Como se mencionó en este post, queremos demostrar diferentes formas de deserialización con Spring Boot y Spring Kafka y simultáneamente ver cómo múltiples consumidores pueden ser balanceados de carga cuando son parte del mismo consumer group.

KafkaExampleApplication.java

@SpringBootApplication
public class KafkaExampleApplication {

    public static void main(String[] args) {
        SpringApplication.run(KafkaExampleApplication.class, args);
    }

    @Autowired
    private KafkaProperties kafkaProperties;

    @Value("${tpd.topic-name}")
    private String topicName;

    // Producer configuration
    // omitted...

    // Consumer configuration

    // If you only need one deserialization type, you only need to set the
    // Consumer configuration properties. Uncomment this and remove all others below.
//    @Bean
//    public Map<String, Object> consumerConfigs() {
//        Map<String, Object> props = new HashMap<>(
//                kafkaProperties.buildConsumerProperties()
//        );
//        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
//                StringDeserializer.class);
//        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
//                JsonDeserializer.class);
//        props.put(ConsumerConfig.GROUP_ID_CONFIG,
//                "tpd-loggers");
//
//        return props;
//    }

    @Bean
    public ConsumerFactory<String, Object> consumerFactory() {
        final JsonDeserializer<Object> jsonDeserializer = new JsonDeserializer<>();
        jsonDeserializer.addTrustedPackages("*");
        return new DefaultKafkaConsumerFactory<>(
                kafkaProperties.buildConsumerProperties(), new StringDeserializer(), jsonDeserializer
        );
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());

        return factory;
    }

    // String Consumer Configuration

    @Bean
    public ConsumerFactory<String, String> stringConsumerFactory() {
        return new DefaultKafkaConsumerFactory<>(
                kafkaProperties.buildConsumerProperties(), new StringDeserializer(), new StringDeserializer()
        );
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerStringContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(stringConsumerFactory());

        return factory;
    }

    // Byte Array Consumer Configuration

    @Bean
    public ConsumerFactory<String, byte[]> byteArrayConsumerFactory() {
        return new DefaultKafkaConsumerFactory<>(
                kafkaProperties.buildConsumerProperties(), new StringDeserializer(), new ByteArrayDeserializer()
        );
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, byte[]> kafkaListenerByteArrayContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, byte[]> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(byteArrayConsumerFactory());
        return factory;
    }
}

Esta configuración puede parecer extensa. La razón es que, para propósitos educativos, hemos repetido tres veces la creación de las instancias ConsumerFactory y KafkaListenerContainerFactory. De esta manera, podemos demostrar estos tres tipos de deserialización asignando uno diferente a cada consumidor.

Los pasos básicos para configurar un consumer son:

1. [Omitido] Configuramos las propiedades del consumer de manera similar a las que hicimos para el producer. Podemos omitir este paso ya que la única configuración que necesitamos es el Group ID, especificado en el archivo de propiedades de Spring Boot, y los deserializadores de clave y valor, que sobrescribiremos mientras creamos las factories personalizadas del consumer y KafkaListener. Si solo necesitas una configuración porque solo necesitas el mismo tipo de deserializador de Clave y Valor (como típicamente sucede en muchas aplicaciones reales), este bloque de código comentado es lo único que necesitas. Ajusta los tipos de deserializadores a los que quieres usar.

2. Crear el ConsumerFactory para ser usado por el KafkaListenerContainerFactory. Creamos tres, cambiando el deserializador de valor en cada caso a 1) un deserializador JSON, 2) un deserializador String, y 3) un deserializador de Byte Array.

   1. Nota que después de crear el JSON Deserializer, incluimos un paso extra para especificar que confiamos en todos los packages. Puedes afinar esto en tu aplicación si quieres. Si no hacemos esto, obtendremos un mensaje de error: java.lang.IllegalArgumentException: The class [] is not in the trusted packages.

3. Construir la factory del contenedor Kafka Listener (una concurrente) usando la Consumer Factory previamente configurada. Nuevamente, hacemos esto tres veces para usar una diferente por instancia.

Recibiendo mensajes con Spring Boot y Kafka en formatos JSON, String, y byte[]

Es hora de mostrar cómo se ven los consumidores de Kafka. Usaremos la anotación @KafkaListener ya que simplifica el proceso y se encarga de la deserialización del tipo Java pasado.

HelloKafkaController.java

@RestController
public class HelloKafkaController {

    private static final Logger logger =
            LoggerFactory.getLogger(HelloKafkaController.class);

    private final KafkaTemplate<String, Object> template;
    private final String topicName;
    private final int messagesPerRequest;
    private CountDownLatch latch;

    public HelloKafkaController(
            final KafkaTemplate<String, Object> template,
            @Value("${tpd.topic-name}") final String topicName,
            @Value("${tpd.messages-per-request}") final int messagesPerRequest) {
        this.template = template;
        this.topicName = topicName;
        this.messagesPerRequest = messagesPerRequest;
    }

    @GetMapping("/hello")
    public String hello() throws Exception {
        latch = new CountDownLatch(messagesPerRequest);
        IntStream.range(0, messagesPerRequest)
                .forEach(i -> this.template.send(topicName, String.valueOf(i),
                        new PracticalAdvice("A Practical Advice", i))
                );
        latch.await(60, TimeUnit.SECONDS);
        logger.info("All messages received");
        return "Hello Kafka!";
    }

    @KafkaListener(topics = "advice-topic", clientIdPrefix = "json",
            containerFactory = "kafkaListenerContainerFactory")
    public void listenAsObject(ConsumerRecord<String, PracticalAdvice> cr,
                              @Payload PracticalAdvice payload) {
        logger.info("Logger 1 [JSON] received key {}: Type [{}] | Payload: {} | Record: {}", cr.key(),
                typeIdHeader(cr.headers()), payload, cr.toString());
        latch.countDown();
    }

    @KafkaListener(topics = "advice-topic", clientIdPrefix = "string",
            containerFactory = "kafkaListenerStringContainerFactory")
    public void listenasString(ConsumerRecord<String, String> cr,
                              @Payload String payload) {
        logger.info("Logger 2 [String] received key {}: Type [{}] | Payload: {} | Record: {}", cr.key(),
                typeIdHeader(cr.headers()), payload, cr.toString());
        latch.countDown();
    }

    @KafkaListener(topics = "advice-topic", clientIdPrefix = "bytearray",
            containerFactory = "kafkaListenerByteArrayContainerFactory")
    public void listenAsByteArray(ConsumerRecord<String, byte[]> cr,
                                  @Payload byte[] payload) {
        logger.info("Logger 3 [ByteArray] received key {}: Type [{}] | Payload: {} | Record: {}", cr.key(),
                typeIdHeader(cr.headers()), payload, cr.toString());
        latch.countDown();
    }

    private static String typeIdHeader(Headers headers) {
        return StreamSupport.stream(headers.spliterator(), false)
                .filter(header -> header.key().equals("__TypeId__"))
                .findFirst().map(header -> new String(header.value())).orElse("N/A");
    }
}

Hay tres listeners en esta clase. Primero, describamos los parámetros de la anotación @KafkaListener para que podamos entender las diferencias:

• Todos los listeners están consumiendo del mismo topic, advice-topic. Este parámetro es obligatorio.
• El parámetro clientIdPrefix es opcional. Lo estoy usando aquí para que los logs sean más amigables para humanos. Sabrás qué consumidor hace qué por su prefijo de nombre. Kafka agregará un número a este prefijo.
• El parámetro containerFactory es opcional; también puedes confiar en la convención de nombres. Si no lo especificas, buscará un bean con el nombre kafkaListenerContainerFactory, que también es el nombre por defecto usado por Spring Boot cuando auto-configura Kafka. También puedes sobrescribirlo usando el mismo nombre (aunque parece magia para alguien que no conoce la convención). Necesitamos establecerlo explícitamente porque queremos usar uno diferente para cada listener, para poder usar diferentes deserializadores.

Nota que el primer argumento pasado a todos los listeners es el mismo: un ConsumerRecord. El segundo, anotado con @Payload es redundante si usamos el primero. Podemos acceder al payload usando el método value() en ConsumerRecord, pero lo incluí para que puedas ver lo simple que es obtener el payload del mensaje directamente por deserialización inferida.

El Header TypeId en Kafka

El header __TypeId__ es establecido automáticamente por la librería de Kafka por defecto. El método de utilidad typeIdHeader que uso aquí es para propósitos educativos. Obtiene la representación string ya que solo verás un array de bytes en la salida del método toString() de ConsumerRecord. Este header TypeId puede ser útil para deserialización, así que puedes encontrar el tipo al cual mapear los datos. No es necesario para deserialización JSON porque el equipo de Spring proporciona un deserializador incorporado que lee ese header e infiere el tipo del argumento del método.

Ejecutando la aplicación

Todo el código en este post está disponible en GitHub: Kafka and Spring Boot Example. Si lo encuentras útil, ¡por favor dale una estrella!

Ahora que hemos terminado el producer y consumers de Kafka, podemos ejecutar Kafka y la app de Spring Boot:

$ docker-compose up -d
Starting kafka-example_zookeeper_1 ... done
Starting kafka-example_kafka_1     ... done
$ mvn spring-boot:run
...

La app de Spring Boot inicia, y los consumidores se registran en Kafka, que les asigna una partición. Configuramos el topic con tres particiones, así que cada consumidor obtiene una de ellas asignada.

Terminal - Spring Boot App - Kafka Topic Partitions

[Consumer clientId=bytearray-0, groupId=tpd-loggers] Subscribed to topic(s): advice-topic
[Consumer clientId=json-0, groupId=tpd-loggers] Subscribed to topic(s): advice-topic
[Consumer clientId=string-0, groupId=tpd-loggers] Subscribed to topic(s): advice-topic
[...]
[Consumer clientId=bytearray-0, groupId=tpd-loggers] Notifying assignor about the new Assignment(partitions=[advice-topic-0])
[Consumer clientId=string-0, groupId=tpd-loggers] Notifying assignor about the new Assignment(partitions=[advice-topic-2])
[Consumer clientId=json-0, groupId=tpd-loggers] Notifying assignor about the new Assignment(partitions=[advice-topic-1])

Ahora podemos probar una llamada HTTP al servicio. Puedes usar tu navegador o una herramienta de línea de comandos como curl, por ejemplo:

Terminal - Hacer una llamada HTTP a /hello

$ curl localhost:8080/hello

La salida en los logs debería verse así:

Terminal - Spring Boot App - Logs del Consumer para cada tipo de deserializador

[ntainer#1-0-C-1] [...]: Logger 2 [String] received key 0: Type [io.tpd.kafkaexample.PracticalAdvice] | Payload: {"message":"A Practical Advice","identifier":0} | Record: ConsumerRecord(topic = advice-topic, partition = 2, leaderEpoch = 0, offset = 16, CreateTime = 1633682802702, serialized key size = 1, serialized value size = 47, headers = RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [105, 111, 46, 116, 112, 100, 46, 107, 97, 102, 107, 97, 101, 120, 97, 109, 112, 108, 101, 46, 80, 114, 97, 99, 116, 105, 99, 97, 108, 65, 100, 118, 105, 99, 101])], isReadOnly = false), key = 0, value = {"message":"A Practical Advice","identifier":0})
[...]
[nio-8080-exec-1] [...]: All messages received

Explicación

Kafka hashea la clave del mensaje (un identificador string simple) y, basado en eso, divide los mensajes en diferentes particiones. Cada consumidor recibe los mensajes en su partición asignada y los convierte a un objeto Java usando su deserializador. Recuerda, nuestro producer siempre envía valores JSON.

Como puedes ver en los logs, cada deserializador logra hacer su tarea. El consumidor String imprime el mensaje JSON crudo, el Byte Array muestra la representación de bytes de ese String JSON, y el deserializador JSON usa el Java Type Mapper para convertirlo a la clase original, PracticalAdvice. Puedes mirar el ConsumerRecord loggeado, y verás los headers, la partición asignada, el offset, etc.

Así es como puedes enviar y recibir mensajes JSON con Spring Boot y Kafka. Y ahora, puedes usar el mismo ejemplo de código para aprender más sobre cómo funciona Kafka. Sigue leyendo abajo.

Prueba algunos Ejercicios de Kafka

Si eres nuevo en Kafka, puede que quieras probar algunos cambios de código para entender mejor cómo funciona.

Solicitar /hello múltiples veces

Haz algunas solicitudes, luego observa cómo los mensajes se distribuyen a través de las particiones. Los mensajes de Kafka con la misma clave siempre se colocan en las mismas particiones. Esta característica es útil cuando quieres asegurar que todos los mensajes para un usuario dado, proceso, o cualquier lógica con la que estés trabajando sean recibidos por el mismo consumidor en el mismo orden en que fueron producidos, sin importar cuánto balanceo de carga estés haciendo. Esto a menudo se refiere como stickiness, y puedes definir qué propiedad usar para adherirse al mismo procesador.

Como ejemplo práctico, digamos que los consumidores están recibiendo objetos UserUpdatedEvent, y tomando algunas decisiones basadas en eso. Por ejemplo, podrían haber tres workers paralelos que son responsables de la lógica de términos de privacidad del usuario:

1. Almacenar la versión de los términos aceptados para cada usuario en una base de datos local, compartida por todas las instancias del consumidor.
2. Eliminar los datos del usuario cuando los datos del evento indican una eliminación de usuario.

Si procesas el evento del mismo usuario en dos workers paralelos, corres el riesgo de tener condiciones de carrera donde un worker lento procesando el evento 1 intenta actualizar un usuario que ya no existe, porque otro worker procesó más rápido un evento de tipo 2. Por eso el modelado de clave de partición es esencial en aplicaciones usando Kafka. Idealmente, todos los eventos para el mismo usuario en esta aplicación son procesados por el mismo worker.

Reducir el número de particiones

Primero, reinicia Kafka, para que descartes la configuración previa.

Luego, redefine el topic en la aplicación para tener solo dos particiones:

KafkaExampleApplication.java - Reconfigurar el topic

@Bean
public NewTopic adviceTopic() {
    return new NewTopic(topicName, 2, (short) 1);
}

Ahora, vuelve a ejecutar la app y haz una petición HTTP al endpoint /hello.

Terminal - Spring Boot App - Consumidor sin partición asignada

Logger 3 [ByteArray] received key 0: Type [io.tpd.kafkaexample.PracticalAdvice] | Payload: [123, 34, 109, 101, 115, 115, 97, 103, 101, 34, 58, 34, 65, 32, 80, 114, 97, 99, 116, 105, 99, 97, 108, 32, 65, 100, 118, 105, 99, 101, 34, 44, 34, 105, 100, 101, 110, 116, 105, 102, 105, 101, 114, 34, 58, 48, 125] | Record: ConsumerRecord(topic = advice-topic, partition = 0, leaderEpoch = 0, offset = 0, CreateTime = 1633685589028, serialized key size = 1, serialized value size = 47, headers = RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [105, 111, 46, 116, 112, 100, 46, 107, 97, 102, 107, 97, 101, 120, 97, 109, 112, 108, 101, 46, 80, 114, 97, 99, 116, 105, 99, 97, 108, 65, 100, 118, 105, 99, 101])], isReadOnly = false), key = 0, value = [B@344c2b5)
[...]
All messages received

Uno de los consumidores (en este caso, la versión String) no recibe ningún mensaje. En este caso de ejemplo, solo los consumidores JSON y ByteArray están activos. Este comportamiento es esperado ya que no hay más particiones disponibles dentro del mismo consumer group para asignarse al consumidor adicional. Para reproducir este comportamiento, puede que necesites destruir el cluster de Kafka en Docker con docker-compose down -v e iniciarlo nuevamente, para que el topic pueda ser recreado desde Java.

La conclusión importante de este ejemplo práctico es que el número de particiones en tu configuración debería ser igual o mayor al número de workers paralelos que planeas tener en tu aplicación.

Cambiar el Group Identifier de un Consumidor

Mantengamos los cambios del caso previo, así que el topic ahora tiene solo dos particiones. Luego, cambia el group ID de uno de nuestros consumidores. Esto conceptualmente lo hará un consumidor independiente, así que ya no compartirá la carga con el otro grupo.

HelloKafkaController - Cambiando Kafka Group Identifier a tpd-loggers-2

@KafkaListener(topics = "advice-topic", clientIdPrefix = "bytearray",
        containerFactory = "kafkaListenerByteArrayContainerFactory",
        groupId = "tpd-loggers-2")
public void listenAsByteArray(ConsumerRecord<String, byte[]> cr,
                              @Payload byte[] payload) {
        logger.info("Logger 3 [ByteArray] received key {}", cr.key());
    latch.countDown();
}

También cambiamos el mensaje loggeado para diferenciarlo fácilmente del resto. Ahora, solo imprime la clave recibida. También necesitamos cambiar el CountDownLatch para que espere el doble del número de mensajes ya que hay dos consumer groups diferentes. Sigue leyendo para una explicación detallada.

HelloKafkaController - Ajustando latch para esperar el doble de mensajes

latch = new CountDownLatch(messagesPerRequest * 2);

¿Por qué necesitamos esto? Como describí al principio de este post, cuando los consumidores pertenecen al mismo Consumer Group, están (conceptualmente) trabajando en la misma tarea. Estamos implementando un mecanismo balanceado de carga en el que workers concurrentes obtienen mensajes de diferentes particiones sin necesitar procesar los mensajes de los demás. Están dividiendo el trabajo a través de workers.

En este ejemplo, también cambié la “tarea conceptual” del último consumidor para entender mejor esto: está haciendo algo diferente, imprimiendo un mensaje diferente en este caso. Como cambiamos el group ID, este consumidor trabaja separadamente de los otros. Kafka le asignará ambas particiones ya que es el único en su grupo. El consumidor Byte Array recibe todos los mensajes ahora. Ve los logs debajo.

Terminal - Spring Boot App - Dos consumer groups diferentes

Logger 3 [ByteArray] received key 0
Logger 3 [ByteArray] received key 2
Logger 3 [ByteArray] received key 5
Logger 3 [ByteArray] received key 6
Logger 2 [String] received key 1: Type [io.tpd.kafkaexample.PracticalAdvice] | Payload: {"message":"A Practical Advice","identifier":1} | Record: ConsumerRecord(topic = advice-topic, partition = 1, leaderEpoch = 0, offset = 0, CreateTime = 1633691293887, serialized key size = 1, serialized value size = 47, headers = RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [105, 111, 46, 116, 112, 100, 46, 107, 97, 102, 107, 97, 101, 120, 97, 109, 112, 108, 101, 46, 80, 114, 97, 99, 116, 105, 99, 97, 108, 65, 100, 118, 105, 99, 101])], isReadOnly = false), key = 1, value = {"message":"A Practical Advice","identifier":1})
[...]
All messages received

Estos ejercicios simples deberían haberte permitido aprender rápidamente estos conceptos esenciales de Kafka en la práctica. Recuerda: si te gustó este post, por favor compártelo, da una estrella al repositorio, o déjame saber tu feedback a través de los comentarios.