diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
index 77d12d37f..c06e4a958 100644
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
+++ b/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) -> {
diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBindingServiceTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBindingServiceTests.java
new file mode 100644
index 000000000..e8153f128
--- /dev/null
+++ b/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 {
+
+	}
+}
diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java
index 8abc02922..246ebd163 100644
--- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java
+++ b/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);
 		}