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 4a6c191e3..5428a4c78 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 @@ -179,6 +179,36 @@ public class StreamBridgeTests { } } + @SuppressWarnings("rawtypes") + @Test + void test_2785() throws Exception { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + EmptyConfiguration.class)).web(WebApplicationType.NONE).run( + "--spring.cloud.stream.source=outputA;outputB", + "--spring.cloud.stream.bindings.outputA-out-0.producer.use-native-encoding=false", + "--spring.cloud.stream.bindings.outputB-out-0.producer.partition-count=1", + "--spring.cloud.stream.bindings.outputC-out-0.producer.use-native-encoding=true", + "--spring.cloud.stream.bindings.outputD-out-0.content-type=text/html", + "--spring.jmx.enabled=false")) { + StreamBridge streamBridge = context.getBean(StreamBridge.class); + Field field = ReflectionUtils.findField(StreamBridge.class, "streamBridgeFunctionCache"); + field.setAccessible(true); + Map functionCache = (Map) field.get(streamBridge); + + streamBridge.send("outputA-out-0", MessageBuilder.withPayload("A").build()); + assertThat(functionCache.size()).isEqualTo(1); + streamBridge.send("foo", MessageBuilder.withPayload("A").build()); + assertThat(functionCache.size()).isEqualTo(1); + streamBridge.send("outputB-out-0", MessageBuilder.withPayload("A").build()); + assertThat(functionCache.size()).isEqualTo(1); + streamBridge.send("outputC-out-0", MessageBuilder.withPayload("A").build()); + assertThat(functionCache.size()).isEqualTo(2); + streamBridge.send("outputD-out-0", MessageBuilder.withPayload("A").build()); + assertThat(functionCache.size()).isEqualTo(3); + } + } + /* * This test verifies that when a partition key expression is set, then scst_partition is always set, even in * concurrent scenarios. 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 01640f72a..8123f8a7f 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 @@ -96,7 +96,7 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi private final BindingService bindingService; - private final Map streamBridgeFunctionCache; + private final Map streamBridgeFunctionCache; private final FunctionInvocationHelper functionInvocationHelper; @@ -187,13 +187,23 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi return messageChannel.send(resultMessage); } + private int hashProducerProperties(ProducerProperties producerProperties, String outputContentType) { + int hash = outputContentType.hashCode() + + Boolean.hashCode(producerProperties.isUseNativeEncoding()) + + Boolean.hashCode(producerProperties.isPartitioned()) + + producerProperties.getPartitionCount(); + + return hash; + } + private synchronized FunctionInvocationWrapper getStreamBridgeFunction(String outputContentType, ProducerProperties producerProperties) { - if (StringUtils.hasText(outputContentType) && this.streamBridgeFunctionCache.containsKey(outputContentType)) { - return this.streamBridgeFunctionCache.get(outputContentType); + int streamBridgeFunctionKey = this.hashProducerProperties(producerProperties, outputContentType); + if (this.streamBridgeFunctionCache.containsKey(streamBridgeFunctionKey)) { + return this.streamBridgeFunctionCache.get(streamBridgeFunctionKey); } else { FunctionInvocationWrapper functionToInvoke = this.functionCatalog.lookup(STREAM_BRIDGE_FUNC_NAME, outputContentType.toString()); - this.streamBridgeFunctionCache.put(outputContentType, functionToInvoke); + this.streamBridgeFunctionCache.put(streamBridgeFunctionKey, functionToInvoke); functionToInvoke.setSkipOutputConversion(producerProperties.isUseNativeEncoding()); return functionToInvoke; }