From f4f6cb509d53d6f802f436295ac0a9f1d5889b9c Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Wed, 9 Sep 2015 15:31:19 -0400 Subject: [PATCH] Add per-binding configuration for partitioning - consumer and producer partitioning properties are defined per-binding - unit tests - Kafka binding fixes for partitioning --- .../kafka/KafkaMessageChannelBinder.java | 1 + .../stream/binder/kafka/KafkaTestBinder.java | 2 +- .../binder/MessageChannelBinderSupport.java | 1 + .../stream/binding/ChannelBindingService.java | 36 ++--- .../config/ChannelBindingProperties.java | 148 +++++++++++++++++- .../partitioning/PartitionedConsumerTest.java | 92 +++++++++++ .../partitioning/PartitionedProducerTest.java | 90 +++++++++++ .../partitioned-consumer-test.properties | 5 + .../partitioned-producer-test.properties | 4 + 9 files changed, 349 insertions(+), 30 deletions(-) create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java create mode 100644 spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/partitioned-consumer-test.properties create mode 100644 spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/partitioned-producer-test.properties diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 05c4c1757..f87b45288 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -391,6 +391,7 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport { retryTemplate.setBackOffPolicy(backOffPolicy); retryOperations = retryTemplate; } + super.afterPropertiesSet(); } /** diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java index 3b4413331..4939c6159 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java @@ -56,10 +56,10 @@ public class KafkaTestBinder extends AbstractTestBinder PRODUCER_BATCHING_BASIC_PROPERTIES = new HashSet( diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java index e907cbf56..c3b83e0a5 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java @@ -23,7 +23,8 @@ import org.springframework.util.Assert; import org.springframework.util.StringUtils; /** - * Handles the binding of input/output channels by delegating to an underlying {@link Binder}. + * Handles the binding of input/output channels by delegating to an underlying + * {@link Binder}. * * @author Mark Fisher * @author Dave Syer @@ -35,59 +36,50 @@ public class ChannelBindingService { private ChannelBindingProperties channelBindingProperties; - public ChannelBindingService(ChannelBindingProperties channelBindingProperties, - Binder binder) { + public ChannelBindingService(ChannelBindingProperties channelBindingProperties, Binder binder) { this.channelBindingProperties = channelBindingProperties; this.binder = binder; } - public void bindConsumer(MessageChannel inputChannel, String inputChannelName) { - String channelBindingTarget = this.channelBindingProperties - .getBindingPath(inputChannelName); + String channelBindingTarget = this.channelBindingProperties.getBindingDestination(inputChannelName); if (isChannelPubSub(channelBindingTarget)) { - this.binder.bindPubSubConsumer(removePrefix(channelBindingTarget), inputChannel, - this.channelBindingProperties.getConsumerProperties()); + this.binder.bindPubSubConsumer(removePrefix(channelBindingTarget), + inputChannel, this.channelBindingProperties.getConsumerProperties(inputChannelName)); } else { this.binder.bindConsumer(channelBindingTarget, inputChannel, - this.channelBindingProperties.getConsumerProperties()); + this.channelBindingProperties.getConsumerProperties(inputChannelName)); } } public void bindProducer(MessageChannel outputChannel, String outputChannelName) { - String channelBindingTarget = this.channelBindingProperties - .getBindingPath(outputChannelName); + String channelBindingTarget = this.channelBindingProperties.getBindingDestination(outputChannelName); if (isChannelPubSub(channelBindingTarget)) { - this.binder.bindPubSubProducer(removePrefix(channelBindingTarget), outputChannel, - this.channelBindingProperties.getProducerProperties()); + this.binder.bindPubSubProducer(removePrefix(channelBindingTarget), + outputChannel, this.channelBindingProperties.getProducerProperties(outputChannelName)); } else { this.binder.bindProducer(channelBindingTarget, outputChannel, - this.channelBindingProperties.getProducerProperties()); + this.channelBindingProperties.getProducerProperties(outputChannelName)); } } private boolean isChannelPubSub(String bindingTarget) { - Assert.isTrue(StringUtils.hasText(bindingTarget), - "Binding target should not be empty/null."); + Assert.isTrue(StringUtils.hasText(bindingTarget), "Binding target should not be empty/null."); return bindingTarget.startsWith("topic:"); } private String removePrefix(String bindingTarget) { - Assert.isTrue(StringUtils.hasText(bindingTarget), - "Binding target should not be empty/null."); + Assert.isTrue(StringUtils.hasText(bindingTarget), "Binding target should not be empty/null."); return bindingTarget.substring(bindingTarget.indexOf(":") + 1); } public void unbindConsumers(String inputChannelName) { - this.binder.unbindConsumers - (inputChannelName); + this.binder.unbindConsumers(inputChannelName); } public void unbindProducers(String outputChannelName) { this.binder.unbindProducers(outputChannelName); } - - } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingProperties.java index 82668942b..a0453982a 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingProperties.java @@ -20,7 +20,9 @@ import java.util.Map; import java.util.Properties; import java.util.TreeMap; +import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.cloud.stream.binder.BinderProperties; import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.annotation.JsonInclude.Include; @@ -33,7 +35,16 @@ import com.fasterxml.jackson.annotation.JsonInclude.Include; @JsonInclude(Include.NON_DEFAULT) public class ChannelBindingProperties { - public static final String PATH = "path"; + public static final String DESTINATION = "destination"; + + public static final String PARTITION_COUNT = "partitionCount"; + + public static final String PARTITIONED = "partitioned"; + + @Value("${INSTANCE_INDEX:${CF_INSTANCE_INDEX:0}}") + private int instanceIndex; + + private int instanceCount = 1; private Properties consumerProperties = new Properties(); @@ -41,7 +52,7 @@ public class ChannelBindingProperties { private Map bindings = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); - public Properties getConsumerProperties() { + private Properties getConsumerProperties() { return this.consumerProperties; } @@ -65,7 +76,24 @@ public class ChannelBindingProperties { this.bindings = bindings; } - public String getBindingPath(String channelName) { + public int getInstanceIndex() { + return instanceIndex; + } + + public void setInstanceIndex(int instanceIndex) { + this.instanceIndex = instanceIndex; + } + + public int getInstanceCount() { + return instanceCount; + } + + public void setInstanceCount(int instanceCount) { + this.instanceCount = instanceCount; + } + + + public String getBindingDestination(String channelName) { Object binding = bindings.get(channelName); // we may shortcut directly to the path if (binding != null) { @@ -74,7 +102,7 @@ public class ChannelBindingProperties { } else if (binding instanceof Map) { Map bindingProperties = (Map) binding; - Object bindingPath = bindingProperties.get(PATH); + Object bindingPath = bindingProperties.get(DESTINATION); if (bindingPath != null) { return bindingPath.toString(); } @@ -84,8 +112,114 @@ public class ChannelBindingProperties { return channelName; } - public String getTapChannelName(String channelName) { - return "tap:" + getBindingPath(channelName); + public int getPartitionCount(String channelName) { + Object binding = bindings.get(channelName); + // we may shortcut directly to the path + if (binding instanceof Map) { + try { + Map bindingProperties = (Map) binding; + Object bindingPath = bindingProperties.get(PARTITION_COUNT); + if (bindingPath != null) { + return Integer.parseInt(bindingPath.toString()); + } + } catch (NumberFormatException e) { + // ignore and just return 1 + } + } + return 1; } - + + public boolean isPartitionedConsumer(String channelName) { + Object binding = bindings.get(channelName); + // if the setting is just a target shortcut + if (binding == null || binding instanceof String) { + return false; + } + else if (binding instanceof Map) { + Map bindingProperties = (Map) binding; + Object bindingPath = bindingProperties.get(PARTITIONED); + if (bindingPath != null) { + return Boolean.valueOf(bindingPath.toString()); + } + } + // just return the channel name if not found + return false; + } + + public boolean isPartitionedProducer(String channelName) { + Object binding = bindings.get(channelName); + // if the setting is just a target shortcut + if (binding == null || binding instanceof String) { + return false; + } + else if (binding instanceof Map) { + Map bindingProperties = (Map) binding; + return bindingProperties.get(BinderProperties.PARTITION_KEY_EXPRESSION) != null + || bindingProperties.get(BinderProperties.PARTITION_KEY_EXTRACTOR_CLASS) != null; + } + return false; + } + + /** + * Merge general properties provided by 'spring.cloud.stream.consumerProperties.*' with individual binding + * properties supplied via binders. + * + * @param inputChannelName the input channel name + * @return merged consumer properties + */ + public Properties getConsumerProperties(String inputChannelName) { + if (isPartitionedConsumer(inputChannelName)) { + Properties channelConsumerProperties = new Properties(); + channelConsumerProperties.putAll(consumerProperties); + channelConsumerProperties.setProperty(BinderProperties.COUNT, + Integer.toString(getInstanceCount())); + channelConsumerProperties.setProperty(BinderProperties.PARTITION_INDEX, + Integer.toString(getInstanceIndex())); + return channelConsumerProperties; + } + else { + return getConsumerProperties(); + } + } + + /** + * Merge general properties provided by 'spring.cloud.stream.producerProperties.*' with individual binding + * properties supplied via binders. + * + * @param outputChannelName the output channel name + * @return merged producer properties + */ + public Properties getProducerProperties(String outputChannelName) { + if (isPartitionedProducer(outputChannelName)) { + Properties channelProducerProperties = new Properties(); + channelProducerProperties.putAll(this.producerProperties); + channelProducerProperties.setProperty(BinderProperties.NEXT_MODULE_COUNT, + Integer.toString(getPartitionCount(outputChannelName))); + copyChannelBindingProperty(outputChannelName, channelProducerProperties, + BinderProperties.PARTITION_KEY_EXPRESSION); + copyChannelBindingProperty(outputChannelName, channelProducerProperties, + BinderProperties.PARTITION_KEY_EXTRACTOR_CLASS); + copyChannelBindingProperty(outputChannelName, channelProducerProperties, + BinderProperties.PARTITION_SELECTOR_CLASS); + copyChannelBindingProperty(outputChannelName, channelProducerProperties, + BinderProperties.PARTITION_SELECTOR_EXPRESSION); + return channelProducerProperties; + } + else { + return this.producerProperties; + } + } + + @SuppressWarnings("unchecked") + private void copyChannelBindingProperty(String outputChannelName, Properties targetProperties, String propertyName) { + Map channelBindingProperties = (Map) bindings.get(outputChannelName); + if (null != channelBindingProperties && channelBindingProperties.containsKey(propertyName)) { + targetProperties.setProperty(propertyName, (String) channelBindingProperties.get(propertyName)); + } + } + + public String getTapChannelName(String channelName) { + return "tap:" + getBindingDestination(channelName); + } + } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java new file mode 100644 index 000000000..9f5b68e80 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java @@ -0,0 +1,92 @@ +/* + * Copyright 2015 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 + * + * http://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.partitioning; + +import static org.hamcrest.core.IsEqual.equalTo; +import static org.hamcrest.text.IsEqualIgnoringCase.equalToIgnoringCase; +import static org.mockito.Matchers.eq; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoMoreInteractions; + +import java.util.Properties; + +import org.junit.Assert; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; +import org.mockito.ArgumentMatcher; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.SpringApplicationConfiguration; +import org.springframework.cloud.stream.annotation.Bindings; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.binder.Binder; +import org.springframework.cloud.stream.binder.BinderProperties; +import org.springframework.cloud.stream.messaging.Sink; +import org.springframework.cloud.stream.messaging.Source; +import org.springframework.cloud.stream.utils.MockBinderConfiguration; +import org.springframework.context.annotation.Import; +import org.springframework.context.annotation.PropertySource; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Marius Bogoevici + */ +@RunWith(SpringJUnit4ClassRunner.class) +@SpringApplicationConfiguration(PartitionedConsumerTest.TestSink.class) + +public class PartitionedConsumerTest { + + @SuppressWarnings("rawtypes") + @Autowired + private Binder binder; + + @Autowired @Bindings(TestSink.class) + private Sink testSource; + + @Test + @SuppressWarnings("unchecked") + public void testBindingPartitionedConsumer() { + ArgumentCaptor argumentCaptor = ArgumentCaptor.forClass(Properties.class); + verify(binder).bindConsumer(eq("partIn"), eq(testSource.input()), argumentCaptor.capture()); + Assert.assertThat(argumentCaptor.getValue().getProperty(BinderProperties.PARTITION_INDEX), equalTo("0")); + Assert.assertThat(argumentCaptor.getValue().getProperty(BinderProperties.COUNT), + equalTo("2")); + verifyNoMoreInteractions(binder); + } + + + @EnableBinding(Sink.class) + @EnableAutoConfiguration + @Import(MockBinderConfiguration.class) + @PropertySource("classpath:/org/springframework/cloud/stream/binder/partitioned-consumer-test.properties") + public static class TestSink { + + } + + class PropertiesArgumentMatcher extends ArgumentMatcher { + @Override + public boolean matches(Object argument) { + if (!(argument instanceof Properties)) { + return false; + } + return true; + } + } + +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java new file mode 100644 index 000000000..b21d52fa5 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java @@ -0,0 +1,90 @@ +/* + * Copyright 2015 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 + * + * http://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.partitioning; + +import static org.hamcrest.core.IsEqual.equalTo; +import static org.mockito.Matchers.eq; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoMoreInteractions; + +import java.util.Properties; + +import org.junit.Assert; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; +import org.mockito.ArgumentMatcher; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.SpringApplicationConfiguration; +import org.springframework.cloud.stream.annotation.Bindings; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.binder.Binder; +import org.springframework.cloud.stream.binder.BinderProperties; +import org.springframework.cloud.stream.messaging.Source; +import org.springframework.cloud.stream.utils.MockBinderConfiguration; +import org.springframework.context.annotation.Import; +import org.springframework.context.annotation.PropertySource; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Marius Bogoevici + */ +@RunWith(SpringJUnit4ClassRunner.class) +@SpringApplicationConfiguration(PartitionedProducerTest.TestSource.class) + +public class PartitionedProducerTest { + + @SuppressWarnings("rawtypes") + @Autowired + private Binder binder; + + @Autowired @Bindings(TestSource.class) + private Source testSource; + + @Test + @SuppressWarnings("unchecked") + public void testBindingPartitionedProducer() { + ArgumentCaptor argumentCaptor = ArgumentCaptor.forClass(Properties.class); + verify(binder).bindProducer(eq("partOut"), eq(testSource.output()), argumentCaptor.capture()); + Assert.assertThat(argumentCaptor.getValue().getProperty(BinderProperties.NEXT_MODULE_COUNT), equalTo("3")); + Assert.assertThat(argumentCaptor.getValue().getProperty(BinderProperties.PARTITION_KEY_EXPRESSION), + equalTo("payload")); + verifyNoMoreInteractions(binder); + } + + + @EnableBinding(Source.class) + @EnableAutoConfiguration + @Import(MockBinderConfiguration.class) + @PropertySource("classpath:/org/springframework/cloud/stream/binder/partitioned-producer-test.properties") + public static class TestSource { + + } + + class PropertiesArgumentMatcher extends ArgumentMatcher { + @Override + public boolean matches(Object argument) { + if (!(argument instanceof Properties)) { + return false; + } + return true; + } + } + +} diff --git a/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/partitioned-consumer-test.properties b/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/partitioned-consumer-test.properties new file mode 100644 index 000000000..c79a9ce74 --- /dev/null +++ b/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/partitioned-consumer-test.properties @@ -0,0 +1,5 @@ +spring.cloud.stream.bindings.input.destination=partIn +spring.cloud.stream.bindings.input.partitioned=true +spring.cloud.stream.instanceCount=2 +spring.cloud.stream.instanceIndex=0 + diff --git a/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/partitioned-producer-test.properties b/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/partitioned-producer-test.properties new file mode 100644 index 000000000..87d462a3a --- /dev/null +++ b/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/partitioned-producer-test.properties @@ -0,0 +1,4 @@ +spring.cloud.stream.bindings.output.destination=partOut +spring.cloud.stream.bindings.output.partitionKeyExpression=payload +spring.cloud.stream.bindings.output.partitionCount=3 +