GH-3257: Preserve configured bindings when StreamBridge evicts a channel

core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java

@@ -644,6 +644,32 @@ void bindingsAreRemovedWithCache() {
 		}
 	}
 
+	@Test
+	void configuredBindingsAreNotRemovedWithCache() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
+			.getCompleteConfiguration(InterceptorConfiguration.class))
+			.web(WebApplicationType.NONE).run(
+				"--spring.jmx.enabled=false",
+				"--spring.cloud.stream.dynamic-destination-cache-size=1",
+				"--spring.cloud.stream.bindings.foo-out-0.destination=fooDestination"
+			)) {
+			StreamBridge bridge = context.getBean(StreamBridge.class);
+			BindingServiceProperties bindingServiceProp = context.getBean(BindingServiceProperties.class);
+
+			bridge.send("foo-out-0", "hello foo");
+			bridge.send("a", "hello a");
+			bridge.send("b", "hello b");
+
+			// foo-out-0 was configured by the application, so cache eviction must leave
+			// its properties alone; otherwise the destination silently reverts to the
+			// binding name the next time the binding is resolved.
+			assertThat(bindingServiceProp.getBindings()).containsKey("foo-out-0");
+			assertThat(bindingServiceProp.getBindings().get("foo-out-0").getDestination())
+				.isEqualTo("fooDestination");
+			assertThat(bindingServiceProp.getBindingDestination("foo-out-0")).isEqualTo("fooDestination");
+		}
+	}
+
 	@Test
 	void withInterceptorsRegisteredOnlyOnOutputChannel() throws InterruptedException {
 		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration

core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java

@@ -19,6 +19,8 @@
 import java.lang.reflect.Type;
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -147,6 +149,11 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
 		this.applicationContext = applicationContext;
 		this.bindingServiceProperties = bindingServiceProperties;
 		this.destinationBindingCallback = destinationBindingCallback;
+		// Binding names present before any dynamic destination is resolved are the ones
+		// the application configured itself. Evicting a channel must not discard their
+		// configuration, only that of destinations this bridge created on demand.
+		Set<String> configuredBindingNames = new TreeSet<>(String.CASE_INSENSITIVE_ORDER);
+		configuredBindingNames.addAll(bindingServiceProperties.getBindings().keySet());
 		this.channelCache = new LinkedHashMap<String, MessageChannel>() {
 			@Override
 			protected boolean removeEldestEntry(Map.Entry<String, MessageChannel> eldest) {
@@ -155,7 +162,9 @@ protected boolean removeEldestEntry(Map.Entry<String, MessageChannel> eldest) {
 					if (logger.isDebugEnabled()) {
 						logger.debug("Removing message channel from cache " + eldest.getKey());
 					}
-					bindingServiceProperties.getBindings().remove(eldest.getKey());
+					if (!configuredBindingNames.contains(eldest.getKey())) {
+						bindingServiceProperties.getBindings().remove(eldest.getKey());
+					}
 					bindingService.unbindProducers(eldest.getKey());
 				}
 				return remove;