Tutorial SpringBoot: Creación de Producers

KafkaTemplate para enviar string sin y con clave

El KafkaTemplate es una herramienta fundamental en Spring Boot para interactuar con Apache Kafka. Permite enviar mensajes al clúster de Kafka de forma sencilla y eficiente. Al trabajar con strings, es posible enviar mensajes sin clave o asociarles una clave específica, lo cual puede ser crucial para ciertos casos de uso, como la partición de mensajes o el aseguramiento de orden.

Para enviar un mensaje de tipo string sin clave utilizando KafkaTemplate, se puede emplear el método send(String topic, V data). Este método acepta el nombre del topic y el mensaje a enviar. Por ejemplo:

@Autowired
private KafkaTemplate<String, String> kafkaTemplate;

public void enviarMensajeSinClave(String mensaje) {
    kafkaTemplate.send("mi-topico", mensaje);
}

En este ejemplo, el mensaje se enviará al topic "mi-topico" y se distribuirá entre las particiones disponibles de forma rotativa o según la configuración del clúster. No se especifica una clave, por lo que el orden de los mensajes no está garantizado entre diferentes consumidores.

Para enviar un mensaje con clave, se utiliza el método send(String topic, K key, V data), donde se añade un parámetro adicional que representa la clave del mensaje. La clave es un elemento esencial para controlar cómo Kafka enruta los mensajes a las particiones. Por ejemplo:

public void enviarMensajeConClave(String clave, String mensaje) {
    kafkaTemplate.send("mi-topico", clave, mensaje);
}

Al proporcionar una clave, Kafka utilizará un algoritmo de particionado (por defecto, un hash de la clave) para determinar a qué partición se envía el mensaje. Esto garantiza que los mensajes con la misma clave son enviados siempre a la misma partición, manteniendo el orden entre ellos.

Es importante tener en cuenta que al inyectar el KafkaTemplate mediante @Autowired, se utiliza la configuración predeterminada definida en el contexto de Spring Boot. Si es necesario personalizar el comportamiento, se puede configurar un ProducerFactory específico.

Además, el KafkaTemplate proporciona métodos adicionales que permiten mayor control sobre el envío de mensajes. Por ejemplo, es posible especificar la partición de destino:

public void enviarMensajeConClaveYParticion(String clave, String mensaje, int particion) {
    kafkaTemplate.send("mi-topico", particion, clave, mensaje);
}

En este caso, el mensaje se enviará a la partición especificada, independientemente de la clave. Este nivel de control es útil cuando se requiere una distribución específica de los mensajes en el clúster.

También es posible añadir cabeceras al mensaje utilizando la clase ProducerRecord. Esto permite incluir metadatos adicionales que pueden ser utilizados por los consumidores:

public void enviarMensajeConCabeceras(String clave, String mensaje) {
    ProducerRecord<String, String> record = new ProducerRecord<>("mi-topico", clave, mensaje);
    record.headers().add("origen", "sistema-externo".getBytes());
    kafkaTemplate.send(record);
}

Las cabeceras proporcionan flexibilidad para transmitir información adicional sin modificar el cuerpo del mensaje. Esta práctica es común en sistemas distribuidos donde se requiere contexto adicional.

Es fundamental manejar las posibles excepciones al enviar mensajes. El método send devuelve un ListenableFuture, lo que permite registrar callbacks para manejar tanto el éxito como los posibles errores:

public void enviarMensajeAsync(String clave, String mensaje) {
    ListenableFuture<SendResult<String, String>> futuro = kafkaTemplate.send("mi-topico", clave, mensaje);
    futuro.addCallback(resultado -> {
        // Éxito en el envío
        System.out.println("Mensaje enviado al topic " + resultado.getRecordMetadata().topic());
    }, ex -> {
        // Error en el envío
        System.err.println("Error al enviar mensaje: " + ex.getMessage());
    });
}

El uso de callbacks permite una gestión asíncrona de los envíos, mejorando el rendimiento de la aplicación y permitiendo reaccionar ante errores de forma eficiente.

Para mejorar la fiabilidad y rendimiento al enviar mensajes, es recomendable ajustar parámetros de configuración como el tamaño del buffer, el número de reintentos o el acks. Estas configuraciones se pueden establecer en el archivo application.properties o mediante una clase de configuración:

spring.kafka.producer.retries=3
spring.kafka.producer.acks=all
spring.kafka.producer.batch-size=16384
spring.kafka.producer.linger-ms=1

Estos parámetros controlan aspectos clave del productor de Kafka. Por ejemplo, establecer acks=all asegura que todos los nodos réplicas confirman la recepción del mensaje antes de considerarlo enviado, aumentando la durabilidad.

KafkaTemplate para enviar JSON y objetos sin usar Schema Registry

El KafkaTemplate de Spring Boot permite enviar mensajes de tipo JSON y objetos serializados sin la necesidad de utilizar un Schema Registry. Para lograr esto, es fundamental configurar adecuadamente los serializadores en el productor. Al enviar objetos Java, se puede aprovechar la integración de Spring con Jackson para realizar la conversión a JSON de forma automática.

Primero, es necesario definir una configuración personalizada para el ProducerFactory que especifique el uso del JsonSerializer. A continuación se muestra cómo crear un ProducerFactory para producir mensajes con claves de tipo String y valores de tipo Object:

@Bean
public ProducerFactory<String, Object> producerFactory() {
    Map<String, Object> configProps = new HashMap<>();
    configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
    return new DefaultKafkaProducerFactory<>(configProps);
}

Con este ProducerFactory, el JsonSerializer se encargará de convertir los objetos Java a JSON antes de enviarlos a Kafka. Luego, se configura el KafkaTemplate utilizando este ProducerFactory:

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

Una vez configurado el KafkaTemplate, se puede enviar un objeto Java directamente. Por ejemplo, supongamos que se tiene una clase Usuario:

public class Usuario {
    private String nombre;
    private int edad;

    // Constructores, getters y setters
}

Para enviar una instancia de Usuario, se utiliza el método send del KafkaTemplate:

@Autowired
private KafkaTemplate<String, Object> kafkaTemplate;

public void enviarUsuario(String topic, Usuario usuario) {
    kafkaTemplate.send(topic, usuario);
}

En este caso, el JsonSerializer serializa el objeto Usuario a JSON, y el mensaje se envía al topic especificado. Es importante destacar que no se requiere un Schema Registry, ya que la estructura del objeto se maneja mediante JSON.

Si se desea enviar el mensaje con una clave, se puede utilizar otra sobrecarga del método send:

public void enviarUsuarioConClave(String topic, String clave, Usuario usuario) {
    kafkaTemplate.send(topic, clave, usuario);
}

Al proporcionar una clave, se controla la asignación del mensaje a una partición específica, lo cual puede ser útil para mantener el orden de los mensajes relacionados.

Es posible también configurar el JsonSerializer para que incluya los tipos de clase en el encabezado del mensaje, lo que facilita la deserialización en el consumidor. Para ello, se puede habilitar la propiedad ADD_TYPE_INFO_HEADERS:

@Bean
public ProducerFactory<String, Object> producerFactory() {
    Map<String, Object> configProps = new HashMap<>();
    configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
    JsonSerializer<Object> jsonSerializer = new JsonSerializer<>();
    jsonSerializer.setAddTypeInfo(true);
    return new DefaultKafkaProducerFactory<>(configProps, new StringSerializer(), jsonSerializer);
}

Esta configuración permite que el productor incluya información de tipo en los headers del mensaje, facilitando al consumidor la tarea de deserializarlo al tipo correcto.

Además, se pueden personalizar las propiedades del JsonSerializer para ajustar el comportamiento de la serialización. Por ejemplo, para ignorar propiedades nulas o formatear las fechas de una manera específica:

JsonSerializer<Object> serializer = new JsonSerializer<>();
serializer.setAddTypeInfo(false);
serializer.setNullValueSerializer(new NullValueSerializer());
serializer.setDateFormat(new SimpleDateFormat("yyyy-MM-dd"));

Esta flexibilidad permite adaptar la serialización a las necesidades específicas de la aplicación.

Al utilizar el KafkaTemplate con objetos Java, se simplifica el proceso de envío de mensajes complejos. Al evitar el uso de un Schema Registry, se reduce la complejidad y se facilita el desarrollo, aunque es importante asegurarse de que productores y consumidores estén sincronizados en cuanto a las clases utilizadas.

Para garantizar la compatibilidad entre productores y consumidores, es recomendable compartir las clases comunes en un módulo o dependencia compartida. De esta manera, ambos extremos manejan las mismas estructuras de datos, evitando conflictos durante la deserialización.

El uso de JSON como formato de intercambio es ampliamente aceptado y facilita la interoperabilidad entre diferentes sistemas. Sin embargo, es esencial tener en cuenta consideraciones de seguridad, como la validación y sanitización de los datos antes de procesarlos en el consumidor.

En conclusión, el KafkaTemplate de Spring Boot permite enviar mensajes de tipo JSON y objetos Java sin la necesidad de un Schema Registry, mediante la correcta configuración de los serializadores y el aprovechamiento de las herramientas de serialización proporcionadas por Jackson y Spring.

Manejo de asincronía, callbacks y errores en productores

El envío asíncrono de mensajes es fundamental para lograr una comunicación eficiente y escalable en aplicaciones que utilizan Apache Kafka. En Spring Boot, el KafkaTemplate permite enviar mensajes de forma asíncrona, proporcionando mecanismos para manejar callbacks y gestionar errores durante el proceso de envío.

Al enviar un mensaje utilizando el método send, el KafkaTemplate devuelve un objeto ListenableFuture<SendResult<K,V>>. Este futuro permite registrar callbacks que se ejecutan cuando el envío se completa, ya sea de forma exitosa o con una excepción.

Por ejemplo, para enviar un mensaje asíncrono y manejar el resultado:

ListenableFuture<SendResult<String, String>> futuro = kafkaTemplate.send("mi-topico", clave, mensaje);

futuro.addCallback(
    resultado -> {
        System.out.println("Mensaje enviado con éxito a la partición "
            + resultado.getRecordMetadata().partition()
            + " con offset " + resultado.getRecordMetadata().offset());
    },
    ex -> {
        System.err.println("Error al enviar el mensaje: " + ex.getMessage());
        // Lógica adicional de manejo de errores
    }
);

En este ejemplo, se registra un callback que implementa ListenableFutureCallback. El método onSuccess se invoca cuando el mensaje se ha enviado correctamente, proporcionando información detallada mediante el objeto SendResult. El método onFailure se ejecuta si ocurre algún error durante el envío.

Es crucial implementar un manejo adecuado de errores en los productores para garantizar la fiabilidad del sistema. Los errores pueden deberse a diversas causas, como fallos en la conexión con el clúster de Kafka, problemas de serialización o excepciones en el propio mensaje.

Para mejorar la tolerancia a fallos, se pueden configurar propiedades del productor que permitan reintentos automáticos. Por ejemplo, ajustando el número de retries y el backoff entre intentos:

spring.kafka.producer.retries=5
spring.kafka.producer.retry-backoff-ms=1000

Con esta configuración, el productor intentará reenviar el mensaje hasta cinco veces, con un intervalo de 1000 milisegundos entre intentos. Esto ayuda a manejar errores transitorios, como fallos temporales de red.

Además, es posible establecer un ErrorHandler personalizado para manejar excepciones no controladas. En la configuración del productor, se puede definir un ProducerListener:

@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
    KafkaTemplate<String, String> template = new KafkaTemplate<>(producerFactory());
    template.setProducerListener(new ProducerListener<>() {
        @Override
        public void onError(ProducerRecord<String, String> record, Exception exception) {
            System.err.println("Error al producir mensaje: " + exception.getMessage());
            // Lógica de notificación o registro de errores
        }
    });
    return template;
}

Este ProducerListener global captura todos los errores ocurridos durante el envío de mensajes, permitiendo implementar acciones como logging detallado o notificaciones a sistemas externos.

La gestión de la asincronía en productores también implica considerar el orden y la consistencia de los mensajes. Si es necesario garantizar el orden, se debe utilizar la misma clave para que los mensajes se dirijan a la misma partición, ya que Kafka asegura el orden dentro de una partición.

Para manejar situaciones en las que el orden es crítico y se requiere confirmación del envío antes de proceder, se puede utilizar el método get del futuro, aunque esto cambia la operación a síncrona:

try {
    SendResult<String, String> resultado = kafkaTemplate.send("mi-topico", clave, mensaje).get();
    System.out.println("Mensaje enviado a partición " + resultado.getRecordMetadata().partition());
} catch (InterruptedException | ExecutionException e) {
    System.err.println("Error al enviar mensaje: " + e.getMessage());
    // Manejo de la excepción
}

Sin embargo, es importante usar este enfoque con precaución, ya que bloquear el hilo puede afectar el rendimiento general del sistema.

Para aplicaciones que requieren transacciones, Spring Kafka soporta transacciones de productor, lo que permite enviar múltiples mensajes como una única unidad atómica. La configuración de transacciones se realiza ajustando las propiedades del productor:

@Bean
public ProducerFactory<String, String> producerFactory() {
    Map<String, Object> configProps = new HashMap<>();
    configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    configProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "prod-1");
    return new DefaultKafkaProducerFactory<>(configProps);
}

@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
    KafkaTemplate<String, String> template = new KafkaTemplate<>(producerFactory());
    template.setTransactionIdPrefix("tx-");
    return template;
}

Una vez configurado, se pueden enviar mensajes dentro de una transacción:

kafkaTemplate.executeInTransaction(operations -> {
    operations.send("mi-topico", clave1, mensaje1);
    operations.send("mi-topico", clave2, mensaje2);
    return true;
});

Si ocurre una excepción dentro del bloque transaccional, todos los mensajes enviados se revertirán, asegurando la consistencia de los datos.

Otra consideración importante es el manejo de timeouts. Por defecto, el productor tiene un tiempo máximo para completar el envío antes de lanzar una excepción. Se puede ajustar el delivery timeout para acomodar situaciones específicas:

spring.kafka.producer.properties.delivery.timeout.ms=30000

Ajustar este valor permite controlar cuánto tiempo espera el productor antes de abandonar el intento de envío, lo cual es útil en entornos con latencias variables.

Es recomendable también implementar estrategias de reconexión en caso de fallos persistentes. Utilizar herramientas como Resilience4j o incorporar lógica de circuit breaker puede mejorar la robustez de la aplicación.

Finalmente, es esencial monitorizar y registrar adecuadamente los eventos relacionados con la producción de mensajes. Utilizar sistemas de logging estructurado y herramientas de observabilidad facilita la detección y resolución de incidencias, contribuyendo a un mantenimiento más eficiente del sistema.

Configuración personalizada de Producers con ProducerFactory

En Spring Boot, el ProducerFactory es el componente encargado de producir instancias de productores de Kafka basados en las propiedades de configuración proporcionadas. Al personalizar el ProducerFactory, es posible ajustar diversos parámetros del productor para adaptarlo a necesidades específicas, como optimizar el rendimiento, mejorar la seguridad o integrar serializadores personalizados.

Para crear un ProducerFactory personalizado, se define un @Bean que devuelve una instancia de ProducerFactory<K, V>. En este contexto, se puede configurar detalladamente las propiedades del productor mediante un mapa de configuración. Por ejemplo:

@Bean
public ProducerFactory<String, String> producerFactory() {
    Map<String, Object> configuracion = new HashMap<>();
    configuracion.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    configuracion.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    configuracion.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    configuracion.put(ProducerConfig.ACKS_CONFIG, "all");
    configuracion.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
    configuracion.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
    return new DefaultKafkaProducerFactory<>(configuracion);
}

En este ejemplo, se configuran propiedades clave del productor como los bootstrap servers, los serializadores de clave y valor, el nivel de acks y el tipo de compresión. Ajustar el valor de acks a "all" garantiza que el líder y todos los réplicas hayan recibido el mensaje antes de considerarlo enviado, aumentando así la fiabilidad.

Además, al personalizar el ProducerFactory, se pueden incorporar interceptores o particionadores personalizados que modifican el comportamiento del productor. Por ejemplo, para utilizar un particionador que distribuya los mensajes según una lógica específica:

configuracion.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MiParticionadorPersonalizado.class);

Donde MiParticionadorPersonalizado es una clase que implementa la interfaz Partitioner de Kafka, permitiendo definir la estrategia de particionado.

Es posible también configurar propiedades relacionadas con el rendimiento, como el tamaño de los lotes (batch size) y el tiempo de espera para acumular mensajes (linger.ms):

configuracion.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
configuracion.put(ProducerConfig.LINGER_MS_CONFIG, 5);

Incrementar el batch.size y el linger.ms puede mejorar el rendimiento al permitir que el productor envíe mensajes en lotes más grandes, pero puede introducir una latencia adicional.

Para incorporar un serializador personalizado en el productor, se puede configurar las clases de serialización de la siguiente manera:

configuracion.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, MiClaveSerializer.class);
configuracion.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, MiValorSerializer.class);

Donde MiClaveSerializer y MiValorSerializer son clases que implementan la interfaz Serializer de Kafka, proporcionando una forma personalizada de serializar las claves y los valores de los mensajes.

Una vez definido el ProducerFactory personalizado, se puede crear un KafkaTemplate que lo utilice:

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

Si se desean manejar distintos tipos de mensajes con diferentes configuraciones, es posible definir múltiples ProducerFactory y KafkaTemplate. Por ejemplo, para enviar mensajes con valores de tipo Evento:

@Bean
public ProducerFactory<String, Evento> producerFactoryEvento() {
    Map<String, Object> configuracion = new HashMap<>(producerConfigs());
    configuracion.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
    return new DefaultKafkaProducerFactory<>(configuracion);
}

@Bean
public KafkaTemplate<String, Evento> kafkaTemplateEvento() {
    return new KafkaTemplate<>(producerFactoryEvento());
}

En este caso, se utiliza un JsonSerializer para serializar objetos de la clase Evento a formato JSON, facilitando la interoperabilidad con sistemas que consumen mensajes en este formato.

Para mejorar la seguridad de las comunicaciones con el clúster de Kafka, se pueden configurar parámetros relacionados con la autenticación SSL o SASL. Por ejemplo, para habilitar la comunicación cifrada mediante SSL:

configuracion.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
configuracion.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "/ruta/al/truststore.jks");
configuracion.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "mi_contraseña");
configuracion.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, "/ruta/al/keystore.jks");
configuracion.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, "mi_contraseña");

Estas configuraciones permiten que el productor establezca conexiones seguras con el clúster de Kafka, lo cual es esencial en entornos corporativos donde la confidencialidad de la información es crítica.

Es posible también ajustar el comportamiento del productor en cuanto a la compresión de los mensajes para optimizar el uso del ancho de banda:

configuracion.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "gzip");

Al habilitar la compresión GZIP, los mensajes se comprimen antes de ser enviados, lo que reduce el tamaño de los datos transmitidos a costa de un mayor uso de CPU.

Otro aspecto importante es configurar el timeout de solicitud, que determina cuánto tiempo espera el productor por una respuesta del clúster:

configuracion.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 15000);

Ajustar este valor es útil en entornos con latencias variables, evitando que el productor falle prematuramente en situaciones de carga elevada.

Para manejar transacciones, se debe asignar un identificador de transacción único al productor:

configuracion.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "mi-transactional-id");

Con esta configuración, el productor puede enviar mensajes de forma transaccional, garantizando la atomicidad y evitando la duplicación de mensajes en caso de fallos.

Además, al utilizar el ProducerFactory, se pueden habilitar métricas de monitorización mediante la configuración de JMX:

configuracion.put(CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG, JmxReporter.class.getName());

Esto permite integrar el productor con herramientas de monitorización, proporcionando visibilidad sobre su rendimiento y facilitando la detección de cuellos de botella.

Si se requiere una lógica personalizada al crear instancias de productores, se puede extender DefaultKafkaProducerFactory y sobrescribir el método createKafkaProducer():

public class MiProducerFactory<K, V> extends DefaultKafkaProducerFactory<K, V> {

    public MiProducerFactory(Map<String, Object> configs) {
        super(configs);
    }

    @Override
    protected KafkaProducer<K, V> createKafkaProducer() {
        KafkaProducer<K, V> producer = super.createKafkaProducer();
        // Inicialización personalizada
        return producer;
    }
}

Al hacer esto, se pueden agregar comportamientos específicos al productor, como registrar eventos personalizados o modificar propiedades después de la creación.

Para configurar globalmente las propiedades del productor en application.properties, y luego cargarlas en el ProducerFactory, se puede utilizar el prefijo spring.kafka.producer:

spring.kafka.producer.bootstrap-servers=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

Y en la configuración de Java, cargarlas de la siguiente manera:

@Bean
public Map<String, Object> producerConfigs() {
    return new HashMap<>(kafkaProperties.buildProducerProperties());
}

Donde kafkaProperties es una instancia de KafkaProperties proporcionada por Spring Boot que inyecta las configuraciones del fichero de propiedades, lo que simplifica la gestión de las mismas.

Es relevante mencionar que al personalizar el ProducerFactory, se tiene un control preciso sobre las configuraciones del productor, lo que es esencial para optimizar el comportamiento del sistema en diferentes escenarios. Ajustar parámetros como el manejo de reintentos, el tamaño de buffers o las políticas de compresión puede tener un impacto significativo en el rendimiento y la fiabilidad de la aplicación.

Es posible utilizar el ProducerFactory para implementar patrones avanzados, como circuit breakers o reintentos inteligentes, integrándolo con librerías como Resilience4j. Esto permite mejorar la resiliencia de la aplicación frente a fallos en el clúster de Kafka u otros componentes de la infraestructura, asegurando una operación continua y confiable.