From 02913cd177ff540e64f591a54a6cb39318b0c660 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 18 Aug 2017 11:49:36 -0400 Subject: [PATCH] GH-169: Use the Actual Partition Count (Producer) Fixes https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/169 If the configured `partitionCount` is less than the physical partition count on an existing topic, the binder emits this message: The `partitionCount` of the producer for topic partJ.0 is 3, smaller than the actual partition count of 8 of the topic. The larger number will be used instead. However, that is not true; the configured partition count is used. Override the configured partition count with the actual partition count. --- .../kafka/KafkaMessageChannelBinder.java | 9 +++++ .../stream/binder/kafka/KafkaBinderTests.java | 38 +++++++++++++------ 2 files changed, 35 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 94b9faeac..b82f4c4e5 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 @@ -166,6 +166,15 @@ public class KafkaMessageChannelBinder extends + producerProperties.getPartitionCount() + ", smaller than the actual partition count of " + partitions.size() + " of 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()); } 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 5071424ff..fd3cbe1d8 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 @@ -97,6 +97,7 @@ import kafka.utils.ZkUtils; * @author Soby Chacko * @author Ilayaperumal Gopinathan * @author Henryk Konsek + * @author Gary Russell */ public abstract class KafkaBinderTests extends PartitionCapableBinderTests, ExtendedProducerProperties> { @@ -975,13 +976,23 @@ public abstract class KafkaBinderTests extends @Test @Override - @SuppressWarnings("unchecked") + @SuppressWarnings({ "unchecked", "rawtypes" }) public void testPartitionedModuleJava() throws Exception { Binder binder = getBinder(); + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + + final ZkClient zkClient; + zkClient = new ZkClient(configurationProperties.getZkConnectionString(), + configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(), + ZKStringSerializer$.MODULE$); + + final ZkUtils zkUtils = new ZkUtils(zkClient, null, false); + invokeCreateTopic(zkUtils, "partJ.0", 8, 1, new Properties()); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); consumerProperties.setConcurrency(2); - consumerProperties.setInstanceCount(3); + consumerProperties.setInstanceCount(4); consumerProperties.setInstanceIndex(0); consumerProperties.setPartitioned(true); consumerProperties.getExtension().setAutoRebalanceEnabled(false); @@ -996,11 +1007,15 @@ public abstract class KafkaBinderTests extends QueueChannel input2 = new QueueChannel(); input2.setBeanName("test.input2J"); Binding input2Binding = binder.bindConsumer("partJ.0", "test", input2, consumerProperties); + consumerProperties.setInstanceIndex(3); + QueueChannel input3 = new QueueChannel(); + input3.setBeanName("test.input3J"); + Binding input3Binding = binder.bindConsumer("partJ.0", "test", input3, consumerProperties); ExtendedProducerProperties producerProperties = createProducerProperties(); producerProperties.setPartitionKeyExtractorClass(PartitionTestSupport.class); producerProperties.setPartitionSelectorClass(PartitionTestSupport.class); - producerProperties.setPartitionCount(3); + producerProperties.setPartitionCount(3); // overridden to 8 on the actual topic DirectChannel output = createBindableChannel("output", createProducerBindingProperties(producerProperties)); output.setBeanName("test.output"); Binding outputBinding = binder.bindProducer("partJ.0", output, producerProperties); @@ -1013,6 +1028,7 @@ public abstract class KafkaBinderTests extends output.send(new GenericMessage<>(2)); output.send(new GenericMessage<>(1)); output.send(new GenericMessage<>(0)); + output.send(new GenericMessage<>(3)); Message receive0 = receive(input0); assertThat(receive0).isNotNull(); @@ -1020,20 +1036,18 @@ public abstract class KafkaBinderTests extends assertThat(receive1).isNotNull(); Message receive2 = receive(input2); assertThat(receive2).isNotNull(); + Message receive3 = receive(input3); + assertThat(receive3).isNotNull(); - if (usesExplicitRouting()) { - assertThat(receive0.getPayload()).isEqualTo(0); - assertThat(receive1.getPayload()).isEqualTo(1); - assertThat(receive2.getPayload()).isEqualTo(2); - } - else { - List> receivedMessages = Arrays.asList(receive0, receive1, receive2); - assertThat(receivedMessages).extracting("payload").containsExactlyInAnyOrder(0, 1, 2); - } + assertThat(receive0.getPayload()).isEqualTo(0); + assertThat(receive1.getPayload()).isEqualTo(1); + assertThat(receive2.getPayload()).isEqualTo(2); + assertThat(receive3.getPayload()).isEqualTo(3); input0Binding.unbind(); input1Binding.unbind(); input2Binding.unbind(); + input3Binding.unbind(); outputBinding.unbind(); }