From 8682417af4b4e4ed67afa56f9cc3b2e21c07924d Mon Sep 17 00:00:00 2001
From: "matthieu.dupuy" <matthieu.dupuy@manomano.com>
Date: Thu, 7 Apr 2022 21:40:12 +0200
Subject: [PATCH] Add setter for Container Customizer

Add setter for Container Customizer on AbstractMessageChannelBinder.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2345
---
 .../docs/src/main/asciidoc/overview.adoc      | 22 +++++++++++++++++++
 .../binder/AbstractMessageChannelBinder.java  | 16 +++++++++++++-
 2 files changed, 37 insertions(+), 1 deletion(-)

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 <<binder-customizer>>):
+====
+[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<C extends ConsumerProperties,
 	 */
 	private final String[] headersToEmbed;
 
-	private final ListenerContainerCustomizer<?> containerCustomizer;
+	private ListenerContainerCustomizer<?> containerCustomizer;
 
 	private final MessageSourceCustomizer<?> sourceCustomizer;
 
@@ -197,6 +197,20 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 				: (ConsumerEndpointCustomizer<MessageProducer>) 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 <L> ListenerContainerCustomizer<L> getContainerCustomizer() {
 		return (ListenerContainerCustomizer<L>) this.containerCustomizer;