GH-2796: Fix partition count related issues (#2807)

If a target middleware (Kafka for ex) topic has more partitions
than what is set on the partition-count producer property, the binder
never updates the original producer partition-count property to the
partition count from the middleware if it is higher than what was given
through the property. Because of this, te PartitionAwareFunctionWrapper
which evaluates and assigns the partition header in Spring Cloud Stream
does not compute the correct partition since the hash operation still
using the original partition-count from the producer binding property.
This commit is addressing this issue.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2796

binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java

@@ -439,6 +439,7 @@ public class KafkaMessageChannelBinder extends
 						+ partitions.size()
 						+ " for the topic. The larger number will be used instead.");
 			}
+			producerProperties.setPartitionCount(partitions.size());
 			List<ChannelInterceptor> interceptors = ((InterceptableChannel) channel)
 					.getInterceptors();
 			interceptors.forEach((interceptor) -> {

binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBindingServiceTests.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2023-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka.integration;
+
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.stream.binder.ProducerProperties;
+import org.springframework.cloud.stream.config.BindingServiceProperties;
+import org.springframework.cloud.stream.function.StreamBridge;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Soby Chacko
+ * @since 4.1.0
+ */
+@ExtendWith(SpringExtension.class)
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = {
+	"spring.cloud.stream.bindings.kafka-binding-service-test.producer.partition-count=2",
+	"spring.cloud.stream.bindings.kafka-binding-service-test.producer.partition-key-expression=headers['partitionKey']"})
+@DirtiesContext
+@EmbeddedKafka(topics = "kafka-binding-service-test", controlledShutdown = true, partitions = 4,
+	bootstrapServersProperty = "spring.kafka.bootstrap-servers")
+public class KafkaBindingServiceTests {
+
+	@Autowired
+	private ApplicationContext context;
+
+	@Test
+	void testKafkaBinderOverridesBindingPartitionCountToTopicPartitionsIfHigher() {
+		final StreamBridge streamBridge = this.context.getBean(StreamBridge.class);
+		GenericMessage<String> message = new GenericMessage<>("foo", Map.of("partitionKey", "key1"));
+		streamBridge.send("kafka-binding-service-test", message);
+		BindingServiceProperties bindingServiceProperties = context.getBean(BindingServiceProperties.class);
+		ProducerProperties producerProperties = bindingServiceProperties.getProducerProperties("kafka-binding-service-test");
+		// Verify that the partition count on the binding is updated to the topic partition count
+		// since the topic was created with more partitions than partition-count on the producer binding.
+		assertThat(producerProperties.getPartitionCount()).isEqualTo(4);
+	}
+
+	@EnableAutoConfiguration
+	@Configuration
+	public static class Config {
+
+	}
+}

core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java

@@ -293,6 +293,14 @@ public class BindingService {
 		validate(producerProperties);
 		Binding<T> binding = doBindProducer(output, bindingTarget, binder,
 				producerProperties);
+		// If the downstream binder modified the partition count in the extended producer properties
+		// based on the higher number of partitions provisioned on the target middleware, update that
+		// in the original producer properties.
+		ProducerProperties originalProducerProperties = this.bindingServiceProperties
+			.getProducerProperties(outputName);
+		if (originalProducerProperties.getPartitionCount() < producerProperties.getPartitionCount()) {
+			originalProducerProperties.setPartitionCount(producerProperties.getPartitionCount());
+		}
 		if (cache) {
 			this.producerBindings.put(outputName, binding);
 		}