Fecha de publicación: 2023-11-27
Configurar un productor y un consumidor es solo el inicio. Operar Kafka en un entorno real implica gestionar topics, particiones, consumer groups, monitoreo y recuperación ante fallos. Este artículo cubre las operaciones más comunes y las buenas prácticas que todo equipo necesita conocer.
Herramientas de línea de comandos
Kafka incluye una serie de scripts de consola que permiten administrar el clúster sin necesidad de interfaces gráficas. Son la forma más directa de inspeccionar y modificar el estado del sistema.
bash
1# Crear un topic con 3 particiones y factor de replicación 2
2kafka-topics.sh --create \
3 --bootstrap-server localhost:9092 \
4 --topic orders \
5 --partitions 3 \
6 --replication-factor 2
7
8# Listar todos los topics
9kafka-topics.sh --list --bootstrap-server localhost:9092
10
11# Describir un topic (ver particiones, réplicas, ISR)
12kafka-topics.sh --describe \
13 --bootstrap-server localhost:9092 \
14 --topic orders
15
16# Aumentar particiones (no se pueden reducir!)
17kafka-topics.sh --alter \
18 --bootstrap-server localhost:9092 \
19 --topic orders \
20 --partitions 6Consumer Groups: la pieza clave del escalamiento
Los consumer groups son el mecanismo que permite a Kafka escalar el consumo horizontalmente. La regla fundamental es que cada partición solo puede ser consumida por un consumidor del mismo grupo al mismo tiempo. Esto implica lo siguiente:
- Si tienes 3 particiones y 3 consumidores, cada uno lee una partición.
- Si tienes 3 particiones y 5 consumidores, 2 estarán ociosos en todo momento.
- Si un consumidor muere, Kafka realiza un rebalanceo automático y redistribuye las particiones entre los consumidores restantes.
bash
1# Listar consumer groups
2kafka-consumer-groups.sh --list \
3 --bootstrap-server localhost:9092
4
5# Ver el estado de un consumer group (lag, offsets, etc.)
6kafka-consumer-groups.sh --describe \
7 --bootstrap-server localhost:9092 \
8 --group order-processor
9
10# Resetear offsets (útil para reprocesar mensajes)
11kafka-consumer-groups.sh --reset-offsets \
12 --bootstrap-server localhost:9092 \
13 --group order-processor \
14 --topic orders \
15 --to-earliest \
16 --execute
17
18# Resetear a una fecha específica
19kafka-consumer-groups.sh --reset-offsets \
20 --bootstrap-server localhost:9092 \
21 --group order-processor \
22 --topic orders \
23 --to-datetime 2024-01-15T00:00:00.000 \
24 --executeMonitoreo: métricas que debes vigilar
Operar Kafka sin monitoreo es operar a ciegas. Estas son las métricas más importantes que debes tener en tu dashboard:
- Consumer Lag: La métrica más importante. Es la diferencia entre el último offset producido y el último offset consumido. Un lag creciente significa que los consumidores no pueden seguir el ritmo de producción.
- Under-replicated partitions: Particiones cuyas réplicas no están sincronizadas. Indica problemas de salud del clúster.
- Request rate y latencia: Tanto del productor como del consumidor. Permiten detectar cuellos de botella antes de que se conviertan en incidentes.
- Disk usage: Kafka almacena mensajes en disco. Si el disco se llena, el broker deja de funcionar.
Retención de mensajes
Por defecto, Kafka retiene los mensajes durante 7 días. Este comportamiento se puede ajustar por topic según las necesidades del negocio, ya sea por tiempo o por tamaño en disco.
bash
1# Configurar retención a 7 días para un topic
2kafka-configs.sh --alter \
3 --bootstrap-server localhost:9092 \
4 --entity-type topics \
5 --entity-name orders \
6 --add-config retention.ms=604800000
7
8# Configurar retención por tamaño (1GB)
9kafka-configs.sh --alter \
10 --bootstrap-server localhost:9092 \
11 --entity-type topics \
12 --entity-name orders \
13 --add-config retention.bytes=1073741824
14
15# Compactación de logs (mantener último valor por key)
16kafka-configs.sh --alter \
17 --bootstrap-server localhost:9092 \
18 --entity-type topics \
19 --entity-name user-profiles \
20 --add-config cleanup.policy=compactLa compactación de logs es especialmente útil para topics que representan el estado actual de una entidad (como perfiles de usuario). En lugar de retener todos los eventos históricos, Kafka mantiene solo el mensaje más reciente por clave.
Configuración del productor en Spring Boot para producción
La configuración por defecto del productor de Kafka prioriza la velocidad por sobre la garantía de entrega. En producción, es fundamental ajustar estos parámetros para evitar pérdida de mensajes.
kotlin
1@Configuration
2class KafkaProducerConfig {
3 @Bean
4 fun producerFactory(): ProducerFactory<String, String> {
5 val config = mapOf(
6 ProducerConfig.BOOTSTRAP_SERVERS_CONFIG to "localhost:9092",
7 ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG to StringSerializer::class.java,
8 ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG to StringSerializer::class.java,
9 // Garantizar entrega
10 ProducerConfig.ACKS_CONFIG to "all",
11 ProducerConfig.RETRIES_CONFIG to 3,
12 ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG to true,
13 // Rendimiento
14 ProducerConfig.BATCH_SIZE_CONFIG to 16384,
15 ProducerConfig.LINGER_MS_CONFIG to 5,
16 ProducerConfig.COMPRESSION_TYPE_CONFIG to "snappy"
17 )
18 return DefaultKafkaProducerFactory(config)
19 }
20}El parámetro
acks=all exige que todas las réplicas del líder confirmen la escritura antes de considerar el mensaje como entregado. Combinado con la idempotencia, garantiza exactamente una entrega incluso ante reintentos por fallos de red.Configuración del consumidor para producción
El consumidor requiere atención especial en dos aspectos: el commit de offsets y la concurrencia. Desactivar el auto-commit te da control total sobre cuándo se marca un mensaje como procesado.
kotlin
1@Configuration
2class KafkaConsumerConfig {
3 @Bean
4 fun consumerFactory(): ConsumerFactory<String, String> {
5 val config = mapOf(
6 ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to "localhost:9092",
7 ConsumerConfig.GROUP_ID_CONFIG to "order-processor",
8 ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java,
9 ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java,
10 ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to "earliest",
11 ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG to false,
12 ConsumerConfig.MAX_POLL_RECORDS_CONFIG to 100,
13 ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG to 300000
14 )
15 return DefaultKafkaConsumerFactory(config)
16 }
17
18 @Bean
19 fun kafkaListenerContainerFactory(
20 consumerFactory: ConsumerFactory<String, String>
21 ): ConcurrentKafkaListenerContainerFactory<String, String> {
22 val factory = ConcurrentKafkaListenerContainerFactory<String, String>()
23 factory.consumerFactory = consumerFactory
24 factory.containerProperties.ackMode = ContainerProperties.AckMode.MANUAL
25 factory.setConcurrency(3) // Número de consumidores concurrentes
26 return factory
27 }
28}El valor de
MAX_POLL_INTERVAL_MS_CONFIG define cuánto tiempo puede tardar el procesamiento de un lote antes de que Kafka considere al consumidor como caído y dispare un rebalanceo. Ajústalo según la complejidad de tu lógica de negocio.Dead Letter Queue: un plan para los mensajes problemáticos
No todos los mensajes se pueden procesar correctamente. Un mensaje malformado, una dependencia no disponible o un error de lógica pueden hacer que el procesamiento falle de forma repetida. La solución es la Dead Letter Queue (DLQ): un topic especial donde se redirigen los mensajes que no se pudieron procesar después de N reintentos.
kotlin
1@KafkaListener(topics = ["orders"], groupId = "order-processor")
2fun processOrder(
3 record: ConsumerRecord<String, String>,
4 acknowledgment: Acknowledgment
5) {
6 try {
7 val order = objectMapper.readValue(record.value(), Order::class.java)
8 orderService.process(order)
9 acknowledgment.acknowledge()
10 } catch (e: Exception) {
11 logger.error("Error procesando orden: ${record.key()}", e)
12 // Enviar a DLQ después de N reintentos
13 kafkaTemplate.send("orders.dlq", record.key(), record.value())
14 acknowledgment.acknowledge() // Acknowledge para avanzar el offset
15 }
16}Es importante hacer el acknowledge incluso cuando el mensaje va a la DLQ. De lo contrario, el offset no avanza y el consumidor quedará procesando el mismo mensaje indefinidamente, bloqueando la partición completa.
Buenas prácticas: resumen operativo
Estas son las recomendaciones que marcan la diferencia entre un clúster estable y uno que genera incidentes constantemente:
- Define una convención de nombres para topics, por ejemplo:
dominio.evento.version. - No crees topics con demasiadas particiones desde el inicio. Puedes agregar más, pero no reducir.
- Usa
acks=alle idempotencia en el productor para garantizar entrega. - Desactiva el auto-commit en producción y usa commit manual.
- Implementa una DLQ para mensajes problemáticos.
- Monitorea el consumer lag como tu alarma principal.
- Usa compresión (
snappyolz4) para reducir el ancho de banda.
Conclusión
Gestionar Kafka no es difícil si conoces las herramientas y los patrones correctos. La clave está en monitorear el consumer lag como indicador principal de salud, configurar la retención adecuada para cada topic según su naturaleza, y tener un plan concreto para los mensajes que fallan. Con estas bases, tu clúster será predecible, observable y resiliente.
