How to Access Channel Name in Listenercontainercustomizer When Using Spring Cloud Stream Binder Kafka
I Am Using Spring-Cloud-Stream-Binder-Kafka and Have Implemented Stateful Retry Using Defaulterrorhandler, I Want to Register Different Defaulterrorhandler for...
I am using spring-cloud-stream-binder-kafka and have implemented stateful retry using DefaultErrorHandler, I want to register different DefaultErrorHandler for different consumer consuming messages from different Kafka topics, and each DefaultErrorHandler will be pushing messages to dlq on recovery based on spring.cloud.stream.kafka.bindings.channel-in-0.consumer.dlqName configuration.
Now when I register a DefaultErrorHandler using ListenerContainerCustomizer I want to access KafkaConsumerProperties for the current input-channel and retrieve dlq name from it and use it in DefaultErrorHandler, so if I can get input-channel name I can get the consumer properties like
@Bean
public ListenerContainerCustomizer<AbstractMessageListenerContainer> containerCustomizer() {
return (container, dest, group) -> {
KafkaConsumerProperties kafkaConsumerProperties = kafkaExtendedBindingProperties.getExtendedConsumerProperties("channel");
String dlqName = kafkaConsumerProperties.getDlqName();
DeadLetterPublishingRecoverer recoverer =
new DeadLetterPublishingRecoverer(template, (rec, ex) -> new TopicPartition(dlqName, -1));
container.setCommonErrorHandler(new DefaultErrorHandler(recoverer,new FixedBackOff(2000L, 5L)));
};
}