From 3c65104b1dcefd5dd260f43b4eaec1565bc6561c Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Thu, 14 Sep 2023 08:52:14 -0400 Subject: [PATCH] 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 --- .../kafka/KafkaMessageChannelBinder.java | 1 + .../integration/KafkaBindingServiceTests.java | 72 +++++++++++++++++++ .../cloud/stream/binding/BindingService.java | 8 +++ 3 files changed, 81 insertions(+) create mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBindingServiceTests.java 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 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 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 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); }