From 1116dd25c8c9e9d4878c7a6ea6413eb4b8303b15 Mon Sep 17 00:00:00 2001 From: kdelay Date: Sun, 11 Oct 2026 17:05:49 +0900 Subject: [PATCH] Remove binding properties of evicted channels sent with a binder name When StreamBridge evicts a channel from its cache it removes the binding properties that were created for the dynamic destination. It did so with the cache key, which is ":" when a binder name is given, while BindingServiceProperties keys the entry by the destination alone. For StreamBridge.send(destination, binderName, data) the removal was therefore a no-op and one BindingProperties entry was kept for every destination ever sent to, regardless of dynamic-destination-cache-size. Use the name the channel was created for when removing the properties and when checking whether the binding was configured by the application. Signed-off-by: kdelay --- .../stream/function/StreamBridgeTests.java | 19 +++++++++++++++++++ .../cloud/stream/function/StreamBridge.java | 7 +++++-- 2 files changed, 24 insertions(+), 2 deletions(-) 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 a37fc7b079..e39336b4d5 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 @@ -944,6 +944,25 @@ void dynamicDestinationWithBinderNameDestroyForCacheSize() { } } + @Test + void bindingsWithBinderNameAreRemovedWithCache() { + 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.binders.kafka1.type=kafka" + )) { + StreamBridge bridge = context.getBean(StreamBridge.class); + bridge.send("binding1", "kafka1", "hello"); + bridge.send("binding2", "kafka1", "hello"); + bridge.send("binding3", "kafka1", "hello"); + + BindingServiceProperties bindingServiceProp = context.getBean(BindingServiceProperties.class); + assertThat(bindingServiceProp.getBindings()).containsOnlyKeys("binding3"); + } + } + @Test void withIntegrationFlowBecauseMarcinSaidSo() { 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 7d6b7627e0..8964c976e1 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 @@ -160,8 +160,11 @@ protected boolean removeEldestEntry(Map.Entry eldest) { if (logger.isDebugEnabled()) { logger.debug("Removing message channel from cache " + eldest.getKey()); } - if (!configuredBindingNames.contains(eldest.getKey())) { - bindingServiceProperties.getBindings().remove(eldest.getKey()); + // The cache key is prefixed with the binder name when one is used, while the + // binding properties are keyed by the binding name the channel was created for. + String bindingName = ((AbstractMessageChannel) eldest.getValue()).getComponentName(); + if (!configuredBindingNames.contains(bindingName)) { + bindingServiceProperties.getBindings().remove(bindingName); } bindingService.unbindProducers(eldest.getKey()); }