diff --git a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java index 9ee01e8e99..44b377f60a 100644 --- a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java +++ b/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 diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java index b87cbe7785..5f4add3309 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java +++ b/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 configuredBindingNames = new TreeSet<>(String.CASE_INSENSITIVE_ORDER); + configuredBindingNames.addAll(bindingServiceProperties.getBindings().keySet()); this.channelCache = new LinkedHashMap() { @Override protected boolean removeEldestEntry(Map.Entry eldest) { @@ -155,7 +162,9 @@ protected boolean removeEldestEntry(Map.Entry 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;