From da268bb6dd35fedc176597d0223f607e57205698 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 23 Jul 2018 13:37:27 -0400 Subject: [PATCH] GH-309: Use actual partition count Fixes https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/309 If more partitions exist than those configured, use the actual. Resolves #416 --- .../kafka/KafkaMessageChannelBinder.java | 29 ++++++++++++------- .../stream/binder/kafka/KafkaBinderTests.java | 15 ++++++++-- 2 files changed, 32 insertions(+), 12 deletions(-) diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 278829c04..d9d8a5514 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -68,12 +68,14 @@ import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerPro import org.springframework.cloud.stream.binder.kafka.properties.KafkaExtendedBindingProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties; import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; +import org.springframework.cloud.stream.binding.MessageConverterConfigurer.PartitioningInterceptor; import org.springframework.cloud.stream.config.ListenerContainerCustomizer; import org.springframework.cloud.stream.provisioning.ConsumerDestination; import org.springframework.cloud.stream.provisioning.ProducerDestination; import org.springframework.context.Lifecycle; import org.springframework.expression.common.LiteralExpression; import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.channel.ChannelInterceptorAware; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; import org.springframework.integration.kafka.inbound.KafkaMessageSource; @@ -106,6 +108,7 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessagingException; +import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.messaging.support.ErrorMessage; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; @@ -217,6 +220,14 @@ public class KafkaMessageChannelBinder extends protected MessageHandler createProducerMessageHandler(final ProducerDestination destination, ExtendedProducerProperties producerProperties, MessageChannel errorChannel) throws Exception { + throw new IllegalStateException("The abstract binder should not call this method"); + } + + @Override + protected MessageHandler createProducerMessageHandler(final ProducerDestination destination, + ExtendedProducerProperties producerProperties, + MessageChannel channel, MessageChannel errorChannel) + throws Exception { /* * IMPORTANT: With a transactional binder, individual producer properties for Kafka are * ignored; the global binder (spring.cloud.stream.kafka.binder.transaction.producer.*) @@ -236,20 +247,18 @@ public class KafkaMessageChannelBinder extends return partitionsFor; }); this.topicsInUse.put(destination.getName(), new TopicInformation(null, partitions)); - if (producerProperties.getPartitionCount() < partitions.size()) { + if (producerProperties.isPartitioned() && producerProperties.getPartitionCount() < partitions.size()) { if (this.logger.isInfoEnabled()) { this.logger.info("The `partitionCount` of the producer for topic " + destination.getName() + " is " + producerProperties.getPartitionCount() + ", smaller than the actual partition count of " - + partitions.size() + " of the topic. The larger number will be used instead."); + + partitions.size() + " for the topic. The larger number will be used instead."); } - /* - * This is dirty; it relies on the fact that we, and the partition interceptor, share a - * hard reference to the producer properties instance. But I don't see another way to fix - * it since the interceptor has already been added to the channel, and we don't have - * access to the channel here; if we did, we could inject the proper partition count - * there. TODO: Consider this when doing the 2.0 binder restructuring. - */ - producerProperties.setPartitionCount(partitions.size()); + List interceptors = ((ChannelInterceptorAware) channel).getChannelInterceptors(); + interceptors.forEach(interceptor -> { + if (interceptor instanceof PartitioningInterceptor) { + ((PartitioningInterceptor) interceptor).setPartitionCount(partitions.size()); + } + }); } KafkaTemplate kafkaTemplate = new KafkaTemplate<>(producerFB); diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 6b53c6e52..9024e0cbc 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -31,6 +31,7 @@ import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import com.fasterxml.jackson.databind.ObjectMapper; @@ -82,6 +83,7 @@ import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerPro import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties; import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; import org.springframework.cloud.stream.binder.kafka.utils.KafkaTopicUtils; +import org.springframework.cloud.stream.binding.MessageConverterConfigurer.PartitioningInterceptor; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.provisioning.ProvisioningException; import org.springframework.context.ApplicationContext; @@ -120,6 +122,7 @@ import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessagingException; import org.springframework.messaging.SubscribableChannel; +import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.GenericMessage; import org.springframework.messaging.support.MessageBuilder; @@ -225,7 +228,7 @@ public class KafkaBinderTests extends return consumerFactory().createConsumer().partitionsFor(topic).size(); } - private void invokeCreateTopic(String topic, int partitions, int replicationFactor) throws Throwable { + private void invokeCreateTopic(String topic, int partitions, int replicationFactor) throws Exception { NewTopic newTopic = new NewTopic(topic, partitions, (short) replicationFactor); @@ -1253,6 +1256,7 @@ public class KafkaBinderTests extends producerProperties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload")); producerProperties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("hashCode()")); producerProperties.setPartitionCount(3); + invokeCreateTopic("output", 6, 1); DirectChannel output = createBindableChannel("output", createProducerBindingProperties(producerProperties)); output.setBeanName("test.output"); @@ -1264,7 +1268,14 @@ public class KafkaBinderTests extends } catch (UnsupportedOperationException ignored) { } - + List interceptors = output.getChannelInterceptors(); + AtomicInteger count = new AtomicInteger(); + interceptors.forEach(interceptor -> { + if (interceptor instanceof PartitioningInterceptor) { + count.set(TestUtils.getPropertyValue(interceptor, "partitionHandler.partitionCount", Integer.class)); + } + }); + assertThat(count.get()).isEqualTo(6); Message message2 = org.springframework.integration.support.MessageBuilder.withPayload(2) .setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "foo") .setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 42)