GH-2997 Fix support for producer's error-handler-definition

Resolves #2997

Author: olegz

binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestContainer.java

@@ -28,9 +28,6 @@
  */
 public final class RabbitTestContainer {
 
-	private RabbitTestContainer() {
-	}
-
 	private static final RabbitMQContainer RABBITMQ;
 	static {
 		String image = "rabbitmq:3.11-management";
@@ -46,6 +43,9 @@ private RabbitTestContainer() {
 		RABBITMQ.start();
 	}
 
+	private RabbitTestContainer() {
+
+	}
 	/**
 	 * Should be called early by test that wants to ensure a shared {@link RabbitMQContainer} is up and running.
 	 */

core/spring-cloud-stream-test-binder/src/test/java/org/springframework/cloud/stream/binder/test/TestChannelBinderTests.java

@@ -43,7 +43,7 @@ public void test() throws Exception {
 				"--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 @@ public String getName() {
 					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 @@ public String getName() {
 					return "sample";
 				}
 			};
-			registerErrorInfrastructure.invoke(binder, destination, "function-in-0");
+			registerErrorInfrastructure.invoke(binder, destination, "function-in-0", false);
 		}
 	}
 

core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java

@@ -305,8 +305,13 @@ public final Binding<MessageChannel> doBindProducer(final String destination,
 				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 @@ protected void afterUnbindConsumer(ConsumerDestination destination, String group
 	 * @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 @@ private SubscribableChannel registerErrorInfrastructure(
 		}
 
 		MessageChannel defaultErrorChannel = null;
-		if (getApplicationContext()
+		if (!errorHandlerDefinitionAvailable && getApplicationContext()
 				.containsBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)) {
 			defaultErrorChannel = getApplicationContext().getBean(
 					IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME,