From 8838d9227e5328393f261da233c32d807c654a16 Mon Sep 17 00:00:00 2001 From: Michailidis Michael Date: Wed, 30 Oct 2019 11:04:19 +0200 Subject: [PATCH] Support consumption from multiple non-native part. Currently, non-native partitioning (e.g. RabbitMQ) require an instance per partition. Add support to allow one instance to consume from multiple partitions. Changed instanceIndex from single valued to List Added a new property to support index list to spawn multiple consumers added test for the index list usage added author to the changed files. Fixes-gh-1232 Using object utils to check if a list is empty Only look at indexes when partitioned Add docs for `instanceIndexList` --- .../main/asciidoc/spring-cloud-stream.adoc | 17 +++-- .../stream/binder/ConsumerProperties.java | 16 +++++ .../cloud/stream/binding/BindingService.java | 33 ++++++++-- .../config/BindingServiceProperties.java | 27 +++++++- .../stream/binding/BindingServiceTests.java | 63 +++++++++++++++++++ 5 files changed, 143 insertions(+), 13 deletions(-) diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index 209381818..839b56a8c 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -1887,18 +1887,23 @@ defaultRetryable:: Whether exceptions thrown by the listener that are not listed in the `retryableExceptions` are retryable. + Default: `true`. -instanceIndex:: -When set to a value greater than equal to zero, it allows customizing the instance index of this consumer (if different from `spring.cloud.stream.instanceIndex`). -When set to a negative value, it defaults to `spring.cloud.stream.instanceIndex`. -See `<>` for more information. -+ -Default: `-1`. instanceCount:: When set to a value greater than equal to zero, it allows customizing the instance count of this consumer (if different from `spring.cloud.stream.instanceCount`). When set to a negative value, it defaults to `spring.cloud.stream.instanceCount`. See `<>` for more information. + Default: `-1`. +instanceIndex:: +When set to a value greater than equal to zero, it allows customizing the instance index of this consumer (if different from `spring.cloud.stream.instanceIndex`). +When set to a negative value, it defaults to `spring.cloud.stream.instanceIndex`. +Ignored if `instanceIndexList` is provided. +See `<>` for more information. ++ +Default: `-1`. +instanceIndexList:: +Used with binders that do not support native partitioning (such as RabbitMQ); allows an application instance to consume from more than one partition. ++ +Default: `-1`. retryableExceptions:: A map of Throwable class names in the key and a boolean in the value. Specify those exceptions (and subclasses) that will or won't be retried. diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java index 4722a3e28..241154497 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java @@ -35,6 +35,7 @@ import org.springframework.messaging.Message; * @author Soby Chacko * @author Oleg Zhurakousky * @author Nicolas Homble + * @author Michael Michailidis */ @JsonInclude(JsonInclude.Include.NON_DEFAULT) public class ConsumerProperties { @@ -74,6 +75,13 @@ public class ConsumerProperties { */ private int instanceIndex = -1; + /** + * When set it will allow the customization of the consumer to spawn a consumer for + * each item in the list. All negative indexes will be discarded. Default: null + * NOTE: This setting will disable the instance-index + */ + private List instanceIndexList; + /** * The number of attempts to process the message (including the first) in the event of * processing failures. This is a RetryTemplate configuration which is provided by the @@ -211,6 +219,14 @@ public class ConsumerProperties { this.instanceIndex = instanceIndex; } + public List getInstanceIndexList() { + return this.instanceIndexList; + } + + public void setInstanceIndexList(List instanceIndexList) { + this.instanceIndexList = instanceIndexList; + } + @Min(value = 1, message = "Max attempts should be greater than zero.") public int getMaxAttempts() { return this.maxAttempts; diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java index 2335cb2ca..124662d0f 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java @@ -53,10 +53,11 @@ import org.springframework.validation.beanvalidation.CustomValidatorBean; * @author Mark Fisher * @author Dave Syer * @author Marius Bogoevici - * @author Ilayaperumal Gopinathan + * @author Ilayaperumal Gopinathan` * @author Gary Russell * @author Janne Valkealahti * @author Soby Chacko + * @author Michael Michailidis */ public class BindingService { @@ -118,13 +119,35 @@ public class BindingService { String[] bindingTargets = StringUtils .commaDelimitedListToStringArray(bindingTarget); for (String target : bindingTargets) { - Binding binding = input instanceof PollableSource + if (!consumerProperties.isPartitioned() || consumerProperties.getInstanceIndexList().isEmpty()) { + Binding binding = input instanceof PollableSource ? doBindPollableConsumer(input, inputName, binder, - consumerProperties, target) + consumerProperties, target) : doBindConsumer(input, inputName, binder, consumerProperties, - target); + target); - bindings.add(binding); + bindings.add(binding); + } + else { + for (Integer index : consumerProperties.getInstanceIndexList()) { + if (index < 0) { + continue; + } + + ConsumerProperties consumerPropertiesTemp = new ExtendedConsumerProperties<>(""); + BeanUtils.copyProperties(consumerProperties, consumerPropertiesTemp); + + consumerPropertiesTemp.setInstanceIndex(index); + + Binding binding = input instanceof PollableSource + ? doBindPollableConsumer(input, inputName, binder, + consumerPropertiesTemp, target) + : doBindConsumer(input, inputName, binder, consumerPropertiesTemp, + target); + + bindings.add(binding); + } + } } } bindings = Collections.unmodifiableCollection(bindings); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java index 45977d0c1..7352b71e9 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java @@ -16,7 +16,9 @@ package org.springframework.cloud.stream.config; +import java.util.ArrayList; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.TreeMap; @@ -49,11 +51,12 @@ import org.springframework.util.Assert; * @author Gary Russell * @author Ilayaperumal Gopinathan * @author Oleg Zhurakousky + * @author Michael Michailidis */ @ConfigurationProperties("spring.cloud.stream") @JsonInclude(Include.NON_DEFAULT) public class BindingServiceProperties - implements ApplicationContextAware, InitializingBean { + implements ApplicationContextAware, InitializingBean { private static final int DEFAULT_BINDING_RETRY_INTERVAL = 30; @@ -66,6 +69,15 @@ public class BindingServiceProperties @Value("${INSTANCE_INDEX:${CF_INSTANCE_INDEX:0}}") private int instanceIndex; + /** + * A list of instance id's from 0 to instanceCount-1. Used for partitioning and with + * Kafka. NOTE: Could also be managed per individual binding + * "spring.cloud.stream.bindings.foo.consumer.instance-index-list" where 'foo' is + * the name of the binding. This setting will override the one set in + * 'spring.cloud.stream.instance-index' + */ + private List instanceIndexList = new ArrayList<>(); + /** * The number of deployed instances of an application. Default: 1. NOTE: Could also be * managed per individual binding @@ -82,7 +94,7 @@ public class BindingServiceProperties * application: 'spring.cloud.stream.bindings.input.contentType=text/plain' */ private Map bindings = new TreeMap<>( - String.CASE_INSENSITIVE_ORDER); + String.CASE_INSENSITIVE_ORDER); /** * Additional per-binder properties (see {@link BinderProperties}) if more then one @@ -146,6 +158,14 @@ public class BindingServiceProperties this.instanceIndex = instanceIndex; } + public List getInstanceIndexList() { + return this.instanceIndexList; + } + + public void setInstanceIndexList(List instanceIndexList) { + this.instanceIndexList = instanceIndexList; + } + public int getInstanceCount() { return this.instanceCount; } @@ -226,6 +246,9 @@ public class BindingServiceProperties if (consumerProperties.getInstanceIndex() < 0) { consumerProperties.setInstanceIndex(this.instanceIndex); } + if (consumerProperties.getInstanceIndexList() == null) { + consumerProperties.setInstanceIndexList(this.instanceIndexList); + } return consumerProperties; } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java index f70c71762..21be8b35b 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java @@ -17,6 +17,7 @@ package org.springframework.cloud.stream.binding; import java.lang.reflect.Field; +import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.HashMap; @@ -103,6 +104,7 @@ import static org.mockito.Mockito.when; * @author Ilayaperumal Gopinathan * @author Janne Valkealahti * @author Soby Chacko + * @author Michael Michailidis */ public class BindingServiceTests { @@ -184,6 +186,67 @@ public class BindingServiceTests { binderFactory.destroy(); } + @SuppressWarnings({ "unchecked", "rawtypes" }) + @Test + public void testMultipleConsumerBindingsFromIndexList() throws Exception { + BindingServiceProperties properties = new BindingServiceProperties(); + Map bindingProperties = new HashMap<>(); + BindingProperties props = new BindingProperties(); + props.setDestination("foo"); + + ConsumerProperties consumer = properties.getConsumerProperties("input"); + consumer.setInstanceIndexList(Arrays.asList(0, 1)); + consumer.setInstanceCount(2); + consumer.setPartitioned(true); + props.setConsumer(consumer); + + final String inputChannelName = "input"; + bindingProperties.put(inputChannelName, props); + + properties.setBindings(bindingProperties); + + DefaultBinderFactory binderFactory = createMockBinderFactory(); + + Binder binder = binderFactory.getBinder("mock", MessageChannel.class); + BindingService service = new BindingService(properties, binderFactory); + MessageChannel inputChannel = new DirectChannel(); + + Binding mockBinding1 = Mockito.mock(Binding.class, "FirstBinding"); + Binding mockBinding2 = Mockito.mock(Binding.class, "SecondBinding"); + + ArgumentCaptor captor = ArgumentCaptor.forClass(ConsumerProperties.class); + + when(binder.bindConsumer(eq("foo"), isNull(), same(inputChannel), + any(ConsumerProperties.class))).thenReturn(mockBinding1).thenReturn(mockBinding2); + + Collection> bindings = service.bindConsumer(inputChannel, + "input"); + assertThat(bindings).hasSize(2); + + Iterator> iterator = bindings.iterator(); + Binding binding1 = iterator.next(); + Binding binding2 = iterator.next(); + + assertThat(binding1).isSameAs(mockBinding1); + assertThat(binding2).isSameAs(mockBinding2); + + service.unbindConsumers("input"); + + verify(binder, times(2)).bindConsumer(eq("foo"), isNull(), same(inputChannel), + captor.capture()); + verify(binding1).unbind(); + verify(binding2).unbind(); + + List allValues = captor.getAllValues(); + + assertThat(allValues.size()).isEqualTo(2); + + assertThat(allValues.get(0).getInstanceIndex()).isEqualTo(0); + assertThat(allValues.get(1).getInstanceIndex()).isEqualTo(1); + + binderFactory.destroy(); + } + @SuppressWarnings({ "unchecked", "rawtypes" }) @Test public void testConsumerBindingWhenMultiplexingIsEnabled() throws Exception {