Spring Kafka listening on infinite loop on error
I have a Spring Kafka listener initialized as
@Bean
public Map<String, Object> consumerConfig() {
final HashMap<String, Object> result = new HashMap<>();
result.put(BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
result.put(GROUP_ID_CONFIG, groupId);
result.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
result.put(VALUE_DESERIALIZER_CLASS_CONFIG, MyKafkaJacksonRulesExecutionResultsDeserializer.class);
return result;
}
@Bean
public ConsumerFactory<Long, MessageResult> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerConfig());
}
@Bean
public ConcurrentKafkaListenerContainerFactory<Long, MessageResult> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<Long, MessageResult> containerFactory = new ConcurrentKafkaListenerContainerFactory<>();
containerFactory.setConsumerFactory(consumerFactory());
containerFactory.setConcurrency(KAFKA_LISTENER_THREADS_COUNT);
containerFactory.getContainerProperties().setPollTimeout(KAFKA_LISTENER_POLL_TIMEOUT);
containerFactory.getContainerProperties().setAckOnError(true);
containerFactory.getContainerProperties().setAckMode(RECORD);
return containerFactory;
}
and used like
@KafkaListener(topics = "${spring.kafka.out-topic}")
public void processSrpResults(MessageResult result) {
The deserializer throws an exception during deserialization which causes an infinite loop because the listener cannot receive a messege.
How can I make the kafka listener fail?
+3
source to share
1 answer
I created a subclass for the deserializer that would throw an exception. Then I used this in my config as a deserializer. Then your processor has to handle null objects.
public class MyErrorHandlingDeserializer extends ExceptionThrowingDeserializer {
@Override
public Object deserialize(String topic, byte[] data) {
try {
return super.deserialize(topic, data);
} catch (Exception e) {
log.error("Problem deserializing data " + new String(data) + " on topic " + topic, e);
return null;
}
}
}
+1
source to share