From 6b26a1f167a84878198cec8cb0e38108e2db29f9 Mon Sep 17 00:00:00 2001 From: adityaanikam Date: Wed, 2 Sep 2026 23:25:23 +0530 Subject: [PATCH] GH-3257: Preserve configured bindings when StreamBridge evicts a channel StreamBridge caps its channel cache at dynamic-destination-cache-size and, on eviction, removes the evicted binding from BindingServiceProperties so that bindings it created for dynamic destinations do not accumulate. That removal is unconditional, so it also discards bindings the application configured itself. Once a configured binding is gone, the next call to BindingServiceProperties#getBindingProperties recreates an empty entry and defaults its destination to the binding name, so messages then go to the binding name instead of the configured destination. An application with more bindings than the cache size hits this as soon as enough distinct bindings are used. Capture the binding names present when StreamBridge is constructed -- those are the ones the application configured -- and skip only those during eviction. Destinations resolved on demand are still removed, so the cleanup added in 07eb699 keeps working and bindingsAreRemovedWithCache passes unchanged. Signed-off-by: adityaanikam --- .../stream/function/StreamBridgeTests.java | 26 +++++++++++++++++++ .../cloud/stream/function/StreamBridge.java | 11 +++++++- 2 files changed, 36 insertions(+), 1 deletion(-) 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;