From 8fc89bc4c073d21839795108ff230daf891b33f3 Mon Sep 17 00:00:00 2001
From: Oleg Zhurakousky <ozhurakousky@vmware.com>
Date: Tue, 24 Sep 2024 17:18:48 +0200
Subject: [PATCH] GH-2997 Fix support for producer's error-handler-definition

Resolves #2997
---
 .../stream/binder/rabbit/RabbitTestContainer.java   |  5 ++++-
 .../stream/binder/test/TestChannelBinderTests.java  |  6 +++---
 .../stream/binder/AbstractMessageChannelBinder.java | 13 +++++++++----
 3 files changed, 16 insertions(+), 8 deletions(-)

diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestContainer.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestContainer.java
index c7de2b242..838ce51c0 100644
--- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestContainer.java
+++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestContainer.java
@@ -25,7 +25,7 @@ import org.testcontainers.containers.RabbitMQContainer;
  *
  * @author Chris Bono
  */
-public class RabbitTestContainer {
+public final class RabbitTestContainer {
 
 	private static final RabbitMQContainer RABBITMQ;
 	static {
@@ -42,6 +42,9 @@ public class RabbitTestContainer {
 		RABBITMQ.start();
 	}
 
+	private RabbitTestContainer() {
+
+	}
 	/**
 	 * Should be called early by test that wants to ensure a shared {@link RabbitMQContainer} is up and running.
 	 */
diff --git a/core/spring-cloud-stream-test-binder/src/test/java/org/springframework/cloud/stream/binder/test/TestChannelBinderTests.java b/core/spring-cloud-stream-test-binder/src/test/java/org/springframework/cloud/stream/binder/test/TestChannelBinderTests.java
index 45a137708..9035cd209 100644
--- a/core/spring-cloud-stream-test-binder/src/test/java/org/springframework/cloud/stream/binder/test/TestChannelBinderTests.java
+++ b/core/spring-cloud-stream-test-binder/src/test/java/org/springframework/cloud/stream/binder/test/TestChannelBinderTests.java
@@ -43,7 +43,7 @@ class TestChannelBinderTests {
 				"--spring.cloud.stream.bindings.function-in-0.destination=input")) {
 			TestChannelBinder binder = context.getBean(TestChannelBinder.class);
 			Method registerErrorInfrastructure = ReflectionUtils
-				.findMethod(TestChannelBinder.class, "registerErrorInfrastructure", ProducerDestination.class, String.class);
+				.findMethod(TestChannelBinder.class, "registerErrorInfrastructure", ProducerDestination.class, String.class, boolean.class);
 			registerErrorInfrastructure.setAccessible(true);
 			ProducerDestination destination = new ProducerDestination() {
 				@Override
@@ -56,7 +56,7 @@ class TestChannelBinderTests {
 					return "sample";
 				}
 			};
-			registerErrorInfrastructure.invoke(binder, destination, "function-in-0");
+			registerErrorInfrastructure.invoke(binder, destination, "function-in-0", false);
 			destination = new ProducerDestination() {
 				@Override
 				public String getNameForPartition(int partition) {
@@ -68,7 +68,7 @@ class TestChannelBinderTests {
 					return "sample";
 				}
 			};
-			registerErrorInfrastructure.invoke(binder, destination, "function-in-0");
+			registerErrorInfrastructure.invoke(binder, destination, "function-in-0", false);
 		}
 	}
 
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 6bd7c5d48..e1ece1cbc 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
@@ -305,8 +305,13 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 				bp = bsp.getBindingProperties(bindingName);
 			}
 
-			SubscribableChannel errorChannel = (bp != null && StringUtils.hasText(bp.getErrorHandlerDefinition())) || producerProperties.isErrorChannelEnabled()
-					? registerErrorInfrastructure(producerDestination, producerProperties.getBindingName()) : null;
+			boolean errorHandlerDefined = bp != null && StringUtils.hasText(bp.getErrorHandlerDefinition());
+			SubscribableChannel errorChannel = errorHandlerDefined || producerProperties.isErrorChannelEnabled()
+					? registerErrorInfrastructure(producerDestination, producerProperties.getBindingName(), errorHandlerDefined)
+							: null;
+
+			String errorChannelName = errorsBaseName(producerDestination, producerProperties.getBindingName());
+			this.subscribeFunctionErrorHandler(errorChannelName, producerProperties.getBindingName());
 
 			producerMessageHandler = createProducerMessageHandler(producerDestination,
 					producerProperties, outputChannel, errorChannel);
@@ -727,7 +732,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 	 * @return the channel.
 	 */
 	private SubscribableChannel registerErrorInfrastructure(
-			ProducerDestination destination, String bindingName) {
+			ProducerDestination destination, String bindingName, boolean errorHandlerDefinitionAvailable) {
 
 		String errorChannelName = errorsBaseName(destination, bindingName);
 		SubscribableChannel errorChannel = new PublishSubscribeChannel();
@@ -751,7 +756,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		}
 
 		MessageChannel defaultErrorChannel = null;
-		if (getApplicationContext()
+		if (!errorHandlerDefinitionAvailable && getApplicationContext()
 				.containsBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)) {
 			defaultErrorChannel = getApplicationContext().getBean(
 					IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME,