Unable to use ListenerContainerWithDlqAndRetryCustomizer

[mattdupuy]

Hi,

I have been trying to use the new ListenerContainerCustomizer "ListenerContainerWithDlqAndRetryCustomizer" for Kafka Binder.
I want to use it to perform a complex retry process.
After following the documentation https://docs.spring.io/spring-cloud-stream-binder-kafka/docs/3.2.2/reference/html/spring-cloud-stream-binder-kafka.html#retry-and-dlq-processing, when I start my application, my customizer is never call, so I have performed some debugging to trying to figure out what is happening.

I have found a strange behavior in the class KafkaMessageChannelBinder, in the method "createConsumerEndpoint", line 748 and line 797.

if (!(customizer instanceof ListenerContainerWithDlqAndRetryCustomizer)
			|| ((ListenerContainerWithDlqAndRetryCustomizer) customizer)
					.retryAndDlqInBinding(destination.getName(), group)) {
		kafkaMessageDrivenChannelAdapter
				.setRetryTemplate(buildRetryTemplate(extendedConsumerProperties));
		kafkaMessageDrivenChannelAdapter
				.setRecoveryCallback(errorInfrastructure.getRecoverer());
}

if (customizer instanceof ListenerContainerWithDlqAndRetryCustomizer) {
	BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> destinationResolver = createDestResolver(
			extendedConsumerProperties.getExtension());
	BackOff createBackOff = extendedConsumerProperties.getMaxAttempts() > 1
			? createBackOff(extendedConsumerProperties)
			: null;
	((ListenerContainerWithDlqAndRetryCustomizer) customizer)
			.configure(messageListenerContainer, destination.getName(), consumerGroup, destinationResolver,
					createBackOff);
}

The code is checking if the customizer is an instance of "ListenerContainerWithDlqAndRetryCustomizer", however, even when the customizer if an instance of "ListenerContainerWithDlqAndRetryCustomizer", the result if false. After investigating, I have found out that the customizer is wrap inside a Anonymous implementaton of ListenerContainerCustomizer, declare inside the DefaultBinderFactory class, line 425.

ListenerContainerCustomizer customizerWrapper = new ListenerContainerCustomizer() {
        @SuppressWarnings("unchecked")
        @Override
        public void configure(Object container, String destinationName, String group) {
	        try {
		        customizerEntry.getValue().configure(container, destinationName, group);
	        }
	        catch (Exception e) {
		        logger.warn("Failed while applying ListenerContainerCustomizer. In situations when multiple "
				        + "binders are used this is expected, since a particular customizer may not be applicable.");
	        }
        }
};
((GenericApplicationContext) binderProducingContext).registerBean(customizerEntry.getKey(),
	ListenerContainerCustomizer.class, () -> customizerWrapper);

Because of this wrapping, even if we declare a bean of type ListenerContainerWithDlqAndRetryCustomizer, it is wrap inside a ListenerContainerCustomizer anonymous implementation, which make the code of the class KafkaMessageChannelBinder always return false on:

customizer instanceof ListenerContainerWithDlqAndRetryCustomizer

Steps to reproduce the behavior:
Follow the documentation inside the documentation:
https://docs.spring.io/spring-cloud-stream-binder-kafka/docs/3.2.2/reference/html/spring-cloud-stream-binder-kafka.html#retry-and-dlq-processing

Version of the framework: 3.2.2
Expected behavior: When a Bean of type ListenerContainerWithDlqAndRetryCustomizer is configured, it use as described in the documentation

Maybe I am doing something wrong in the configuration, but I haven't found out yet.

Thanks for looking into it.

[garyrussell]

It looks like that code is trying to copy the customizer(s) from the parent context and it knows nothing about the sub-interface.

If you only have one binder, don't give it a name - named binders are intended for when multiple binders are present.

If you must give it a name:

When using named binders, you can use a BinderCustomizer to add a custom ListenerContainerCustomizer to the binder instead of declaring it as a @Bean.

See https://docs.spring.io/spring-cloud-stream/docs/3.2.2/reference/html/spring-cloud-stream.html#binder-customizer

[mattdupuy]

Thanks for the feedback @garyrussell .
I have indeed 2 binders (Rabbit and Kafka), the purpose of the application is to move event from rabbit to kafka and then process them from kafka.
I will look into this BinderCustomizer to configure my ListenerContainerWithDlqAndRetryCustomizer and let you know the result.

If this solve the issue, maybe we could improve the documentation of https://docs.spring.io/spring-cloud-stream-binder-kafka/docs/3.2.2/reference/html/spring-cloud-stream-binder-kafka.html#retry-and-dlq-processing to reference this sub chapter in case of multi binder.

[sobychacko]

Sure, @mattdupuy. Let us know if that worked for you. You are welcome to submit a PR to improve the documentation.

[mattdupuy]

I have look into using a BinderCustomizer, however I do not find any accessor to provide a custom ListenerContainerCustomizer as it is part of the abstract class AbstractMessageChannelBinder which does not provide any setter and it can only be set by using the super constructor.

Am I missing something ?

[mattdupuy]

Hi again,

I have been able to use the BinderCustomizer to change the ListenerContainerCustomizer using some reflection to bypass the fact that there is not setter for the containerCustomizer attribute:

    @Bean
    public BinderCustomizer binderCustomizer(KafkaListenerContainerWithDlqAndRetryCustomizer customizer) {
        return (binder, binderName) -> {
            if (binder.getClass().isAssignableFrom(KafkaMessageChannelBinder.class)) {
                try {
                    Field nameField = binder.getClass().getSuperclass()
                            .getDeclaredField("containerCustomizer");
                    nameField.setAccessible(true);

                    nameField.set(binder, customizer);
                } catch (NoSuchFieldException | IllegalAccessException e) {
                    throw new RuntimeException("Unable to customize the KafkaMessageChannelBinder to configure the KafkaListenerContainerWithDlqAndRetryCustomizer customizer.", e);
                }
            }
        };
    }

However, I am not a big fan of using Reflection to bypass the normal behavior of a Framework, as there might be a reason for this field to not be accessible. Do you think this is a valid option for this use case? Could we expect a change in the library to make the field overridable ?

[garyrussell]

That's probably why they tried to copy the beans into the binder context; there should be a setter for this.