From bc562e3a773ea72c44d53c5fa801c494efca922b Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Wed, 17 Jan 2018 15:52:06 -0500 Subject: [PATCH] Polished KafkaBinderTests polished KafkaBinderTests to account for changes in core related to partitioning - https://github.com/spring-cloud/spring-cloud-stream/commit/00748985d622e84a25b23d16585d9d35e1be3816 Resolves #272 --- .../cloud/stream/binder/kafka/KafkaBinderTests.java | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) 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 92f047b1e..bcd202a16 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 @@ -84,6 +84,7 @@ import org.springframework.cloud.stream.provisioning.ProvisioningException; import org.springframework.context.ApplicationContext; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.support.GenericApplicationContext; +import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; @@ -580,7 +581,6 @@ public class KafkaBinderTests extends AbstractKafkaTestBinder binder = getBinder(); ExtendedProducerProperties producerProperties = createProducerProperties(); - producerProperties.setPartitionCount(2); producerProperties.getExtension().setHeaderPatterns(new String[]{MessageHeaders.CONTENT_TYPE}); DirectChannel moduleOutputChannel = createBindableChannel("output", @@ -660,7 +660,6 @@ public class KafkaBinderTests extends Binder binder = getBinder(); ExtendedProducerProperties producerProperties = createProducerProperties(); - producerProperties.setPartitionCount(10); BindingProperties producerBindingProperties = createProducerBindingProperties( producerProperties); @@ -719,7 +718,6 @@ public class KafkaBinderTests extends public void testDefaultAutoCommitOnErrorWithDlq() throws Exception { Binder binder = getBinder(); ExtendedProducerProperties producerProperties = createProducerProperties(); - producerProperties.setPartitionCount(10); BindingProperties producerBindingProperties = createProducerBindingProperties( producerProperties); @@ -788,7 +786,6 @@ public class KafkaBinderTests extends Binder binder = getBinder(); ExtendedProducerProperties producerProperties = createProducerProperties(); - producerProperties.setPartitionCount(10); ExtendedConsumerProperties consumerProperties = createConsumerProperties(); consumerProperties.setMaxAttempts(3); consumerProperties.setBackOffInitialInterval(100); @@ -1614,6 +1611,7 @@ public class KafkaBinderTests extends QueueChannel moduleInputChannel = new QueueChannel(); ExtendedProducerProperties producerProperties = createProducerProperties(); producerProperties.setPartitionCount(10); + producerProperties.setPartitionKeyExpression(new LiteralExpression("foo")); DirectChannel moduleOutputChannel = createBindableChannel("output", createProducerBindingProperties(producerProperties));