Crear proyecto con Apache Kafka

Paso 1: datos del proyecto

En start.spring.io generamos un nuevo proyecto con Java, maven y Spring Boot:

Paso 2: seleccionar dependencias

Vamos a crear una aplicación web de Spring con Apache Kafka.

Dependencias:

• Lombok
• Spring Web
• Spring Data JPA
• H2 o MySQL
• Apache Kafka

Paso 3: configurar logging

En desarrollo, puede ser interesante habilitar los logs de Kafka para ver la conexión y el envío y recepción de la información hacia los topics.

En application properties:

logging.level.org.apache.kafka=DEBUG

Paso 4: configuración de propiedades para Apache Kafka

Una vez creado y descargado el proyecto, lo abrimos con IntelliJ IDEA o nuestro IDE preferido.

En el archivo application.properties podemos configurar propiedades para personalizar:

• Propiedades generales de kafka
• Propiedades para los Producer
• Propiedades para los Consumer

Incluso es común tener solo Producers en un proyecto y solo Consumers en otro proyecto. Normalmente habrá combinación de ambos en un mismo proyecto ya que en arquitecturas con múltiples aplicaciones Spring Boot se suele utilizar Apache Kafka como middleware para intercambiar información entre las distintas aplicaciones.

Para configurar una aplicación Spring Boot 3 para trabajar con Apache Kafka utilizando Producers, Consumers y Processors, es esencial definir correctamente diversas propiedades en el archivo application.properties. A continuación, se detallan todas las propiedades necesarias, organizadas por categorías, con una explicación de cada una y recomendaciones sobre los valores a asignar.

1. Configuración General de Kafka

Estas propiedades establecen la conexión básica y comportamientos generales de Kafka en la aplicación.

spring.kafka.bootstrap-servers

• Descripción: Especifica la dirección(s) del/los broker(s) de Kafka a los que la aplicación se conectará.
• Valor recomendado: Lista de direcciones y puertos de los brokers, separadas por comas.
• Ejemplo:

  spring.kafka.bootstrap-servers=localhost:9092,localhost:9093

spring.kafka.client-id

• Descripción: Identificador único para el cliente Kafka. Útil para el monitoreo y trazabilidad.
• Valor recomendado: Un identificador descriptivo para la aplicación o componente.
• Ejemplo:

  spring.kafka.client-id=my-kafka-client

spring.kafka.properties.*

• Descripción: Permite configurar propiedades adicionales de Kafka que no están cubiertas por las propiedades de Spring Boot.
• Valor recomendado: Según las necesidades específicas de la aplicación.
• Ejemplo:

  spring.kafka.properties.security.protocol=SASL_SSL
  spring.kafka.properties.sasl.mechanism=PLAIN

2. Configuración de producers (Productores)

Estas propiedades configuran cómo los productores enviarán mensajes a Kafka.

spring.kafka.producer.key-serializer

• Descripción: Clase encargada de serializar las claves de los mensajes.
• Valor recomendado: org.apache.kafka.common.serialization.StringSerializer si las claves son cadenas de texto.
• Ejemplo:

  spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer

spring.kafka.producer.value-serializer

• Descripción: Clase encargada de serializar los valores de los mensajes.
• Valor recomendado: org.apache.kafka.common.serialization.StringSerializer para valores de texto o org.springframework.kafka.support.serializer.JsonSerializer para objetos JSON.
• Ejemplo:

  spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

spring.kafka.producer.acks

• Descripción: Define el nivel de reconocimiento que el productor requiere del broker.
• Valores posibles: 0, 1, all (o -1).
  • 0: No espera reconocimiento.
  • 1: Espera que el líder confirme.
  • all: Espera que todos los réplicas confirmen.
• Valor recomendado: all para mayor durabilidad.
• Ejemplo:

  spring.kafka.producer.acks=all

spring.kafka.producer.retries

• Descripción: Número de reintentos que realizará el productor en caso de fallo al enviar un mensaje.
• Valor recomendado: Depende de la tolerancia a fallos, comúnmente 3.
• Ejemplo:

  spring.kafka.producer.retries=3

spring.kafka.producer.batch-size

• Descripción: Tamaño en bytes del lote de mensajes que el productor enviará en una sola solicitud.
• Valor recomendado: 16384 (16 KB) es un valor predeterminado común.
• Ejemplo:

  spring.kafka.producer.batch-size=16384

spring.kafka.producer.linger-ms

• Descripción: Tiempo en milisegundos que el productor esperará antes de enviar un lote de mensajes.
• Valor recomendado: 1 para baja latencia o mayor para mayor rendimiento.
• Ejemplo:

  spring.kafka.producer.linger-ms=1

spring.kafka.producer.buffer-memory

• Descripción: Cantidad total de memoria disponible para el buffer de mensajes del productor.
• Valor recomendado: 33554432 (32 MB) es un valor predeterminado común.
• Ejemplo:

  spring.kafka.producer.buffer-memory=33554432

3. Configuración de Consumers (Consumidores)

Estas propiedades configuran cómo los consumidores recibirán y procesarán mensajes de Kafka.

spring.kafka.consumer.group-id

• Descripción: Identificador del grupo de consumidores al que pertenece este consumidor.
• Valor recomendado: Nombre descriptivo del grupo, por ejemplo, my-group.
• Ejemplo:

  spring.kafka.consumer.group-id=my-group

spring.kafka.consumer.auto-offset-reset

• Descripción: Define qué hacer cuando no hay un offset inicial o el offset actual ya no es válido.
• Valores posibles: earliest, latest, none.
  • earliest: Comienza a leer desde el inicio del log.
  • latest: Comienza a leer desde el último mensaje.
  • none: Lanza una excepción si no hay offset.
• Valor recomendado: earliest para asegurarse de procesar todos los mensajes.
• Ejemplo:

  spring.kafka.consumer.auto-offset-reset=earliest

spring.kafka.consumer.enable-auto-commit

• Descripción: Habilita o deshabilita el auto-commit de los offsets.
• Valor recomendado: false para gestionar manualmente los commits.
• Ejemplo:

  spring.kafka.consumer.enable-auto-commit=false

spring.kafka.consumer.key-deserializer

• Descripción: Clase encargada de deserializar las claves de los mensajes.
• Valor recomendado: org.apache.kafka.common.serialization.StringDeserializer si las claves son cadenas de texto.
• Ejemplo:

  spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer

spring.kafka.consumer.value-deserializer

• Descripción: Clase encargada de deserializar los valores de los mensajes.
• Valor recomendado: org.apache.kafka.common.serialization.StringDeserializer para valores de texto o org.springframework.kafka.support.serializer.JsonDeserializer para objetos JSON.
• Ejemplo:

  spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer

spring.kafka.consumer.max-poll-records

• Descripción: Número máximo de registros que el consumidor puede obtener en una sola llamada de polling.
• Valor recomendado: 500 es un valor común, ajustable según la carga.
• Ejemplo:

  spring.kafka.consumer.max-poll-records=500

spring.kafka.consumer.session.timeout-ms

• Descripción: Tiempo máximo entre los latidos del consumidor para detectar fallos.
• Valor recomendado: 15000 (15 segundos) es un valor predeterminado común.
• Ejemplo:

  spring.kafka.consumer.session.timeout-ms=15000

4. Configuración de Processors (Procesadores)

Los Processors en Kafka Streams combinan características de Producers y Consumers, además de capacidades de procesamiento de flujos. A continuación, se presentan las propiedades esenciales para configurar Kafka Streams en Spring Boot.

spring.kafka.streams.application-id

• Descripción: Identificador único de la aplicación de Kafka Streams.
• Valor recomendado: Nombre descriptivo de la aplicación, por ejemplo, my-streams-app.
• Ejemplo:

  spring.kafka.streams.application-id=my-streams-app

spring.kafka.streams.properties.default.key.serde

• Descripción: Serde por defecto para las claves en los flujos.
• Valor recomendado: org.apache.kafka.common.serialization.Serdes$StringSerde para cadenas de texto.
• Ejemplo:

  spring.kafka.streams.properties.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde

spring.kafka.streams.properties.default.value.serde

• Descripción: Serde por defecto para los valores en los flujos.
• Valor recomendado: org.apache.kafka.common.serialization.Serdes$StringSerde para cadenas de texto o un Serde adecuado para otros tipos de datos.
• Ejemplo:

  spring.kafka.streams.properties.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde

spring.kafka.streams.commit.interval.ms

• Descripción: Intervalo de tiempo entre commits de offsets en Kafka Streams.
• Valor recomendado: 1000 (1 segundo) es un valor común.
• Ejemplo:

  spring.kafka.streams.commit.interval.ms=1000

spring.kafka.streams.cache.max.bytes.buffering

• Descripción: Tamaño máximo del buffer de caché para operaciones de estado en Kafka Streams.
• Valor recomendado: 10485760 (10 MB) es un valor predeterminado común.
• Ejemplo:

  spring.kafka.streams.cache.max.bytes.buffering=10485760

spring.kafka.streams.processing.guarantee

• Descripción: Define la garantía de procesamiento de la aplicación Kafka Streams.
• Valores posibles: at_least_once o exactly_once.
• Valor recomendado: exactly_once para asegurar la precisión en el procesamiento.
• Ejemplo:

  spring.kafka.streams.processing.guarantee=exactly_once

5. Configuración de Seguridad (Opcional)

Si tu clúster de Kafka requiere seguridad, deberás configurar propiedades adicionales.

spring.kafka.security.protocol

• Descripción: Protocolo de seguridad a usar.
• Valores posibles: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL.
• Valor recomendado: Según la configuración de tu clúster.
• Ejemplo:

  spring.kafka.security.protocol=SASL_SSL

spring.kafka.properties.sasl.mechanism

• Descripción: Mecanismo de autenticación SASL.
• Valores posibles: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, etc.
• Valor recomendado: Según la configuración de tu clúster.
• Ejemplo:

  spring.kafka.properties.sasl.mechanism=PLAIN

spring.kafka.properties.ssl.truststore.location

• Descripción: Ruta al archivo de truststore SSL.
• Valor recomendado: Ruta absoluta o relativa al archivo de truststore.
• Ejemplo:

  spring.kafka.properties.ssl.truststore.location=/path/to/truststore.jks

spring.kafka.properties.ssl.truststore.password

• Descripción: Contraseña para el truststore SSL.
• Valor recomendado: Contraseña segura.
• Ejemplo:

  spring.kafka.properties.ssl.truststore.password=changeit

6. Configuraciones adicionales

spring.kafka.listener.concurrency

• Descripción: Número de hilos concurrentes para los listeners de Kafka.
• Valor recomendado: Según los recursos disponibles y la carga esperada, por ejemplo, 3.
• Ejemplo:

  spring.kafka.listener.concurrency=3

spring.kafka.listener.type

• Descripción: Tipo de listener a utilizar.
• Valores posibles: single, batch.
• Valor recomendado: batch si se procesan mensajes en lotes.
• Ejemplo:

  spring.kafka.listener.type=batch

spring.kafka.template.default-topic

• Descripción: Tema por defecto para el KafkaTemplate si no se especifica otro.
• Valor recomendado: Nombre del tema principal de producción.
• Ejemplo:

  spring.kafka.template.default-topic=my-default-topic

spring.kafka.template.producer-properties.*

• Descripción: Permite configurar propiedades específicas del productor para el KafkaTemplate.
• Valor recomendado: Según las necesidades específicas, por ejemplo, configuraciones de seguridad.
• Ejemplo:

  spring.kafka.template.producer-properties.security.protocol=SASL_SSL

7. Ejemplo completo de application.properties

A continuación, se presenta un ejemplo completo que integra las propiedades mencionadas anteriormente:

# Configuración General
spring.kafka.bootstrap-servers=localhost:9092,localhost:9093
spring.kafka.client-id=my-kafka-client

# Configuración de Producer
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.producer.acks=all
spring.kafka.producer.retries=3
spring.kafka.producer.batch-size=16384
spring.kafka.producer.linger-ms=1
spring.kafka.producer.buffer-memory=33554432

# Configuración de Consumer
spring.kafka.consumer.group-id=my-group
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=false
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.max-poll-records=500
spring.kafka.consumer.session.timeout-ms=15000

# Configuración de Kafka Streams (Processor)
spring.kafka.streams.application-id=my-streams-app
spring.kafka.streams.properties.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.kafka.streams.properties.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.kafka.streams.commit.interval.ms=1000
spring.kafka.streams.cache.max.bytes.buffering=10485760
spring.kafka.streams.processing.guarantee=exactly_once

# Configuración de Seguridad (Opcional)
spring.kafka.security.protocol=SASL_SSL
spring.kafka.properties.sasl.mechanism=PLAIN
spring.kafka.properties.ssl.truststore.location=/path/to/truststore.jks
spring.kafka.properties.ssl.truststore.password=changeit

# Configuraciones Adicionales
spring.kafka.listener.concurrency=3
spring.kafka.listener.type=batch
spring.kafka.template.default-topic=my-default-topic