GH-3242: Key StreamBridge function cache by binding name

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 {
 

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());