diff --git a/binders/kafka-binder/docs/src/main/asciidoc/overview.adoc b/binders/kafka-binder/docs/src/main/asciidoc/overview.adoc index 24a9c31d3..b20ebe01c 100644 --- a/binders/kafka-binder/docs/src/main/asciidoc/overview.adoc +++ b/binders/kafka-binder/docs/src/main/asciidoc/overview.adoc @@ -906,6 +906,28 @@ ListenerContainerWithDlqAndRetryCustomizer cust(KafkaTemplate template) { Now, only a single retry delay needs to be greater than the consumer's `max.poll.interval.ms` property. +When working with several binders, the 'ListenerContainerWithDlqAndRetryCustomizer' bean gets overridden by the 'DefaultBinderFactory'. For the bean +to apply, you need to use a 'BinderCustomizer' to set the container customizer (See <>): +==== +[source, java] +---- +@Bean +public BinderCustomizer binderCustomizer(ListenerContainerWithDlqAndRetryCustomizer containerCustomizer) { + return (binder, binderName) -> { + if (binder instanceof KafkaMessageChannelBinder) { + ((KafkaMessageChannelBinder) binder).setContainerCustomizer(containerCustomizer); + } + else if (binder instanceof KStreamBinder) { + ... + } + else if (binder instanceof RabbitMessageChannelBinder) { + ... + } + }; +} +---- +==== + [[consumer-producer-config-customizer]] === Customizing Consumer and Producer configuration diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index f56dd2194..5525f4b5b 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java @@ -113,7 +113,7 @@ public abstract class AbstractMessageChannelBinder containerCustomizer; + private ListenerContainerCustomizer containerCustomizer; private final MessageSourceCustomizer sourceCustomizer; @@ -197,6 +197,20 @@ public abstract class AbstractMessageChannelBinder) endpointCustomizer; } + /** + * Configure an optional {@link ListenerContainerCustomizer} for further + * configuration of the listener container instance created by the binder. + * @param containerCustomizer the {@link ListenerContainerCustomizer} to use. + */ + @SuppressWarnings("unchecked") + public void setContainerCustomizer(@Nullable ListenerContainerCustomizer containerCustomizer) { + + this.containerCustomizer = + containerCustomizer == null + ? (container, destinationName, group) -> { } + : containerCustomizer; + } + @SuppressWarnings("unchecked") protected ListenerContainerCustomizer getContainerCustomizer() { return (ListenerContainerCustomizer) this.containerCustomizer;