StreamBridge's dynamic destinations should be closed on shutdown.

- if not, we might lose some messages on shutdown especially in case of kafka async mode producing.

Resolves #2835
This commit is contained in:
kurt
2023-10-20 14:39:33 +09:00
committed by Oleg Zhurakousky
parent eebbe7a5bb
commit 3575c9937d
2 changed files with 28 additions and 1 deletions

View File

@@ -639,6 +639,25 @@ public class StreamBridgeTests {
}
}
@Test
void testDynamicDestinationDestroy() {
BindingService bindingService;
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"
)) {
StreamBridge bridge = context.getBean(StreamBridge.class);
bridge.send("a", "hello foo");
bindingService = context.getBean(BindingService.class);
assertThat(bindingService.getProducerBindingNames().length).isEqualTo(1);
assertThat(bindingService.getProducerBindingNames()[0]).isEqualTo("a");
}
assertThat(bindingService.getProducerBindingNames().length).isEqualTo(0);
}
@Test
void testWithIntegrationFlowBecauseMarcinSaidSo() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration

View File

@@ -26,6 +26,7 @@ import java.util.function.Function;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.cloud.function.context.FunctionCatalog;
import org.springframework.cloud.function.context.FunctionRegistration;
@@ -77,7 +78,7 @@ import org.springframework.util.StringUtils;
*
*/
@SuppressWarnings("rawtypes")
public final class StreamBridge implements StreamOperations, SmartInitializingSingleton {
public final class StreamBridge implements StreamOperations, SmartInitializingSingleton, DisposableBean {
private static final String STREAM_BRIDGE_FUNC_NAME = "streamBridge";
@@ -297,4 +298,11 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
this.applicationContext.getBean(GlobalChannelInterceptorProcessor.class);
globalChannelInterceptorProcessor.postProcessAfterInitialization(messageChannel, destinationName);
}
@Override
public void destroy() throws Exception {
channelCache.keySet().forEach(bindingService::unbindProducers);
channelCache.clear();
}
}