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 c18c394178..88aa9f4e5a 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 @@ -47,6 +47,8 @@ import org.springframework.cloud.function.cloudevent.CloudEventMessageBuilder; import org.springframework.cloud.function.cloudevent.CloudEventMessageUtils; import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper; +import org.springframework.cloud.stream.binder.BinderHeaders; +import org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy; import org.springframework.cloud.stream.binder.test.InputDestination; import org.springframework.cloud.stream.binder.test.OutputDestination; import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; @@ -487,6 +489,31 @@ void delayedSend() { } } + // See https://github.com/spring-cloud/spring-cloud-stream/issues/3242 + @Test + void test_3242() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + PartitionKeyExtractorConfiguration.class)).web(WebApplicationType.NONE).run( + "--spring.cloud.stream.source=partitioned;nonPartitioned", + "--spring.cloud.stream.bindings.partitioned-out-0.producer.partition-count=7", + "--spring.cloud.stream.bindings.partitioned-out-0.producer.partition-key-extractor-name=partitionKeyExtractor", + "--spring.cloud.stream.bindings.nonPartitioned-out-0.producer.partition-count=1", + "--spring.jmx.enabled=false")) { + StreamBridge streamBridge = context.getBean(StreamBridge.class); + + streamBridge.send("partitioned-out-0", + MessageBuilder.withPayload("partitioned").setHeader("partitionKey", "key").build()); + streamBridge.send("nonPartitioned-out-0", MessageBuilder.withPayload("nonPartitioned").build()); + + OutputDestination output = context.getBean(OutputDestination.class); + assertThat(output.receive(1000, "partitioned-out-0").getHeaders() + .containsKey(BinderHeaders.PARTITION_HEADER)).isTrue(); + assertThat(output.receive(1000, "nonPartitioned-out-0").getHeaders() + .containsKey(BinderHeaders.PARTITION_HEADER)).isFalse(); + } + } + @Test void withInterceptorsMatchedAgainstAllPatterns() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration @@ -930,6 +957,16 @@ public static class EmptyConfiguration { } + @EnableAutoConfiguration + public static class PartitionKeyExtractorConfiguration { + + @Bean + public PartitionKeyExtractorStrategy partitionKeyExtractor() { + return message -> message.getHeaders().get("partitionKey"); + } + + } + @EnableAutoConfiguration public static class EmptyConfigurationWithCustomConverters { 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 2f3aaf8f06..75880af6f9 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,7 @@ import java.lang.reflect.Type; import java.util.LinkedHashMap; import java.util.Map; +import java.util.Objects; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -196,7 +197,7 @@ public boolean send(String bindingName, @Nullable String binderName, Object data ProducerProperties producerProperties = this.bindingServiceProperties.getProducerProperties(bindingName); MessageChannel messageChannel = this.resolveDestination(bindingName, producerProperties, binderName); - Function functionToInvoke = this.getStreamBridgeFunction(outputContentType.toString(), producerProperties); + Function functionToInvoke = this.getStreamBridgeFunction(bindingName, outputContentType.toString(), producerProperties); if (producerProperties != null && producerProperties.isPartitioned()) { functionToInvoke = new PartitionAwareFunctionWrapper(functionToInvoke, this.applicationContext, producerProperties); @@ -232,21 +233,22 @@ public boolean send(String bindingName, @Nullable String binderName, Object data 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(); - - if (producerProperties.getPartitionKeyExpression() != null && producerProperties.getBindingName() != null) { - hash += producerProperties.getBindingName().hashCode(); - } - - return hash; + private int hashProducerProperties(String bindingName, ProducerProperties producerProperties, String outputContentType) { + /* + * A partitioned binding mutates the cached function by setting the partition enhancer on it, + * so it must never share a cached function with another binding. The binding name is taken + * from the send(..) argument, since ProducerProperties#getBindingName() is only populated for + * binders that are not ExtendedPropertiesBinder (see GH-3242). + */ + return Objects.hash(outputContentType, + producerProperties.isUseNativeEncoding(), + producerProperties.isPartitioned(), + producerProperties.getPartitionCount(), + producerProperties.isPartitioned() ? bindingName : null); } - private FunctionInvocationWrapper getStreamBridgeFunction(String outputContentType, ProducerProperties producerProperties) { - int streamBridgeFunctionKey = this.hashProducerProperties(producerProperties, outputContentType); + private FunctionInvocationWrapper getStreamBridgeFunction(String bindingName, String outputContentType, ProducerProperties producerProperties) { + int streamBridgeFunctionKey = this.hashProducerProperties(bindingName, producerProperties, outputContentType); return this.streamBridgeFunctionCache.computeIfAbsent(streamBridgeFunctionKey, key -> { FunctionInvocationWrapper functionToInvoke = this.functionCatalog.lookup(STREAM_BRIDGE_FUNC_NAME, outputContentType.toString());