From 1cda3acb22ed3ae2aad8e820bb35299b87c3c6d1 Mon Sep 17 00:00:00 2001
From: Soby Chacko <soby.chacko@broadcom.com>
Date: Mon, 22 Jan 2024 17:19:55 -0500
Subject: [PATCH] GH-2883: CCE from custom error channel bean

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2883

When an applicaiton provides a custom errorChannel bean that is not of type
`PublishSubscribeChannel`, the application fails to start with a CCE. This is
because `BindingServiceConfiguration` uses a hard cast on `PublishSubscribeChannel`
without pre-checking the channel type. Fixing this issue.
---
 .../BindingServiceConfigurationTests.java     | 25 ++++++++++++++++++-
 .../config/BindingServiceConfiguration.java   |  6 ++---
 2 files changed, 27 insertions(+), 4 deletions(-)

diff --git a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/BindingServiceConfigurationTests.java b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/BindingServiceConfigurationTests.java
index 618f95eb98..84e31e00de 100644
--- a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/BindingServiceConfigurationTests.java
+++ b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/BindingServiceConfigurationTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2023 the original author or authors.
+ * Copyright 2017-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -26,9 +26,11 @@
 import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.PublishSubscribeChannel;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.messaging.support.GenericMessage;
@@ -37,6 +39,7 @@
 
 /**
  * @author Oleg Zhurakousky
+ * @author Soby Chacko
  *
  */
 class BindingServiceConfigurationTests {
@@ -81,6 +84,17 @@ void valdateImportedConfiguartionHandlerPostProcessing() {
 		}
 	}
 
+	@Test // See: https://github.com/spring-cloud/spring-cloud-stream/issues/2883
+	void customErrorChannelDoesNotThrowExceptions() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+			TestChannelBinderConfiguration
+				.getCompleteConfiguration(CustomErrorChannelConfiguration.class))
+			.web(WebApplicationType.NONE).run()) {
+			DirectChannel channel = context.getBean("errorChannel", DirectChannel.class);
+			assertThat(channel).isNotNull();
+		}
+	}
+
 	@Configuration
 	@Import(ImportedConfiguration.class)
 	public static class RootConfiguration {
@@ -105,4 +119,13 @@ public static class EmptyConfiguration {
 
 	}
 
+	@Configuration
+	public static class CustomErrorChannelConfiguration {
+
+		@Bean
+		public DirectChannel errorChannel() {
+			return new DirectChannel();
+		}
+	}
+
 }
diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java
index 56a6f9f0c7..0c424b6bbe 100644
--- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java
+++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2022 the original author or authors.
+ * Copyright 2015-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -152,8 +152,8 @@ public static BeanPostProcessor globalErrorChannelCustomizer() {
 		return new BeanPostProcessor() {
 			@Override
 			public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
-				if ("errorChannel".equals(beanName)) {
-					((PublishSubscribeChannel) bean).setIgnoreFailures(true);
+				if ("errorChannel".equals(beanName) && bean instanceof PublishSubscribeChannel publishSubscribeChannel) {
+						publishSubscribeChannel.setIgnoreFailures(true);
 				}
 				return bean;
 			}