He estado tratando de hacer un trabajo de POC para Spring Kafka. Específicamente, quería experimentar cuáles son las mejores prácticas en términos de manejo de errores al consumir mensajes dentro de Kafka.

Me pregunto si alguien puede ayudar con:

1. Compartiendo las mejores prácticas sobre lo que los consumidores de Kafka deben hacer cuando hay una falla
2. Ayúdame a comprender cómo funciona AckMode Record y cómo evitar confirmaciones en la cola de compensación de Kafka cuando se lanza una excepción en el método de escucha.

El ejemplo de código para 2 se da a continuación:

Dado que AckMode está configurado en RECORD, que según la documentación :

comete el desplazamiento cuando el oyente regresa después de procesar el registro.

Hubiera pensado que el desplazamiento no se incrementaría si el método de escucha arrojara una excepción. Sin embargo, este no fue el caso cuando lo probé usando la combinación de código/configuración/comando a continuación. El desplazamiento aún se actualiza y el siguiente mensaje continúa procesándose.

Mi configuración:

 private Map<String, Object> producerConfigs() { Map<String, Object> props = new HashMap<>(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.0.1:9092"); props.put(ProducerConfig.RETRIES_CONFIG, 0); props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384); props.put(ProducerConfig.LINGER_MS_CONFIG, 1); props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); return props; } @Bean ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() { ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs())); factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.RECORD); return factory; }

Mi código:

 @Component public class KafkaMessageListener{ @KafkaListener(topicPartitions = {@TopicPartition( topic = "my-replicated-topic", partitionOffsets = @PartitionOffset(partition = "0", initialOffset = "0", relativeToCurrent = "true"))}) public void onReplicatedTopicMessage(ConsumerRecord<Integer, String> data) throws InterruptedException { throw new RuntimeException("Oops!"); }

Comando para verificar el desplazamiento:

 bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group test-group

Estoy usando kafka_2.12-0.10.2.0 y org.springframework.kafka:spring-kafka:1.1.3.RELEASE