From 2fa9cda89c162bdaa4cb439c9c57e5f1f0066dea Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Thu, 18 Feb 2016 10:59:52 -0500 Subject: [PATCH] Durability Configuration and Default Groups Resolves #317 Remove the `durable` binder configuration property Make subscriber groups durable by default Introduce `requiredGroups` property Kafka groups (non-anonymous) now start by default at EARLIEST, which is more appropriate for new stream consumers Addressing PR comments --- .../kafka/KafkaMessageChannelBinder.java | 11 +- .../stream/binder/kafka/KafkaBinderTests.java | 6 +- .../rabbit/RabbitMessageChannelBinder.java | 62 +++++++--- .../RabbitBinderConfigurationProperties.java | 2 +- ...bbitMessageChannelBinderConfiguration.java | 1 + .../binder/rabbit/RabbitBinderTests.java | 5 +- .../redis/RedisMessageChannelBinder.java | 9 +- .../binder/PartitionCapableBinderTests.java | 58 +++++++++ .../cloud/stream/binder/AbstractBinder.java | 13 +-- .../stream/binder/BinderPropertyKeys.java | 11 +- .../DefaultBindingPropertiesAccessor.java | 13 ++- .../stream/config/BindingProperties.java | 19 +-- .../ChannelBindingServiceProperties.java | 110 +++++++++--------- 13 files changed, 205 insertions(+), 115 deletions(-) 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 32e675561..cc09688c1 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 @@ -191,6 +191,7 @@ public class KafkaMessageChannelBinder extends AbstractBinder { private static final Set KAFKA_PRODUCER_PROPERTIES = new SetBuilder() .add(BinderPropertyKeys.MIN_PARTITION_COUNT) + .add(BinderPropertyKeys.REQUIRED_GROUPS) .build(); /** @@ -249,7 +250,7 @@ public class KafkaMessageChannelBinder extends AbstractBinder { private boolean resetOffsets = DEFAULT_RESET_OFFSETS; - private StartOffset startOffset = DEFAULT_START_OFFSET; + private StartOffset startOffset = null; private int zkSessionTimeout = DEFAULT_ZK_SESSION_TIMEOUT; @@ -440,9 +441,13 @@ public class KafkaMessageChannelBinder extends AbstractBinder { // Consumers reset offsets at the latest time by default, which allows them to receive only // messages sent after they've been bound. That behavior can be changed with the // "resetOffsets" and "startOffset" properties. - String consumerGroup = group == null ? "anonymous." + UUID.randomUUID().toString() : group; + boolean anonymous = !StringUtils.hasText(group); + String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID().toString() : group; + // The reference point, if not set explicitly is the latest time for anonymous subscriptions and the + // earliest time for group subscriptions. This allows the latter to receive messages published before the group + // has been created. long referencePoint = this.startOffset != null ? - startOffset.getReferencePoint() : OffsetRequest.LatestTime(); + startOffset.getReferencePoint() : (anonymous ? OffsetRequest.LatestTime() : OffsetRequest.EarliestTime()); return createKafkaConsumer(name, inputChannel, properties, consumerGroup, referencePoint); } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 383d5d8d3..5e3587352 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -17,7 +17,6 @@ package org.springframework.cloud.stream.binder.kafka; import static org.hamcrest.Matchers.equalTo; -import static org.hamcrest.Matchers.is; import static org.hamcrest.Matchers.not; import static org.hamcrest.Matchers.nullValue; import static org.hamcrest.collection.IsCollectionWithSize.hasSize; @@ -340,7 +339,7 @@ public class KafkaBinderTests extends PartitionCapableBinderTests { @Test @SuppressWarnings("unchecked") - public void testDefaultConsumerStartsAtLatest() throws Exception { + public void testDefaultConsumerStartsAtEarliest() throws Exception { KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(new ZookeeperConnect(kafkaTestSupport.getZkConnectString()), kafkaTestSupport.getBrokerAddress(), kafkaTestSupport.getZkConnectString()); GenericApplicationContext context = new GenericApplicationContext(); @@ -357,7 +356,8 @@ public class KafkaBinderTests extends PartitionCapableBinderTests { output.send(new GenericMessage<>(testPayload1.getBytes())); binder.bindConsumer(testTopicName, "startOffsets", input1, properties); Message receivedMessage1 = (Message) receive(input1); - assertThat(receivedMessage1, is(nullValue())); + assertThat(receivedMessage1, not(nullValue())); + assertThat(new String(receivedMessage1.getPayload()), equalTo(testPayload1)); String testPayload2 = "foo-" + UUID.randomUUID().toString(); output.send(new GenericMessage<>(testPayload2.getBytes())); Message receivedMessage2 = (Message) receive(input1); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index e655ce755..a9b983175 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -134,7 +134,6 @@ public class RabbitMessageChannelBinder extends AbstractBinder { private static final String DEAD_LETTER_EXCHANGE = "DLX"; private static final Set RABBIT_CONSUMER_PROPERTIES = new HashSet(Arrays.asList(new String[] { - BinderPropertyKeys.MAX_CONCURRENCY, RabbitPropertiesAccessor.ACK_MODE, RabbitPropertiesAccessor.PREFETCH, @@ -144,7 +143,8 @@ public class RabbitMessageChannelBinder extends AbstractBinder { RabbitPropertiesAccessor.TRANSACTED, RabbitPropertiesAccessor.TX_SIZE, RabbitPropertiesAccessor.AUTO_BIND_DLQ, - RabbitPropertiesAccessor.REPUBLISH_TO_DLQ + RabbitPropertiesAccessor.REPUBLISH_TO_DLQ, + RabbitPropertiesAccessor.DURABLE })); /** @@ -161,7 +161,6 @@ public class RabbitMessageChannelBinder extends AbstractBinder { */ private static final Set SUPPORTED_CONSUMER_PROPERTIES = new SetBuilder() .addAll(SUPPORTED_BASIC_CONSUMER_PROPERTIES) - .add(BinderPropertyKeys.DURABLE) .add(BinderPropertyKeys.CONCURRENCY) .add(BinderPropertyKeys.PARTITION_INDEX) .build(); @@ -175,6 +174,7 @@ public class RabbitMessageChannelBinder extends AbstractBinder { .add(RabbitPropertiesAccessor.PREFIX) .add(RabbitPropertiesAccessor.REQUEST_HEADER_PATTERNS) .add(BinderPropertyKeys.COMPRESS) + .add(BinderPropertyKeys.REQUIRED_GROUPS) .build(); /** @@ -232,6 +232,8 @@ public class RabbitMessageChannelBinder extends AbstractBinder { private volatile int defaultTxSize = DEFAULT_TX_SIZE; + protected volatile boolean defaultDurableSubscription = true; + private volatile String defaultPrefix = DEFAULT_RABBIT_PREFIX; private volatile String[] defaultRequestHeaderPatterns = DEFAULT_REQUEST_HEADER_PATTERNS; @@ -325,6 +327,15 @@ public class RabbitMessageChannelBinder extends AbstractBinder { this.defaultTxSize = defaultTxSize; } + /** + * Set whether subscriptions are durable. + * @param defaultDurableSubscription true for durable (default false). + */ + public void setDefaultDurableSubscription(boolean defaultDurableSubscription) { + this.defaultDurableSubscription = defaultDurableSubscription; + } + + public void setDefaultPrefix(String defaultPrefix) { Assert.notNull(defaultPrefix, "'defaultPrefix' cannot be null"); this.defaultPrefix = defaultPrefix.trim(); @@ -555,26 +566,32 @@ public class RabbitMessageChannelBinder extends AbstractBinder { declareExchange(exchangeName, exchange); AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(rabbitTemplate); endpoint.setExchangeName(exchange.getName()); - String baseQueueName = exchangeName + ".default"; if (partitionKeyExpression == null && !StringUtils.hasText(partitionKeyExtractorClass)) { - Queue queue = new Queue(baseQueueName, true, false, false, queueArgs(properties, baseQueueName)); - declareQueue(baseQueueName, queue); - autoBindDLQ(baseQueueName, baseQueueName, properties); endpoint.setRoutingKey(name); - org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue).to(exchange).with(name); - declareBinding(baseQueueName, binding); } else { endpoint.setExpressionRoutingKey(EXPRESSION_PARSER.parseExpression(buildPartitionRoutingExpression(name))); - // if the stream is partitioned, create one queue for each target partition for the default group - for (int i = 0; i < properties.getNextModuleCount(); i++) { - String partitionSuffix = "-" + i; - String partitionQueueName = baseQueueName + partitionSuffix; - Queue queue = new Queue(partitionQueueName, true, false, false, - queueArgs(properties, partitionQueueName)); - declareQueue(queue.getName(), queue); - autoBindDLQ(baseQueueName, baseQueueName + partitionSuffix, properties); - declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange).with(name + partitionSuffix)); + } + for (String requiredGroupName : properties.getRequiredGroups(defaultRequiredGroups)) { + String baseQueueName = exchangeName + "." + requiredGroupName; + if (partitionKeyExpression == null && !StringUtils.hasText(partitionKeyExtractorClass)) { + Queue queue = new Queue(baseQueueName, true, false, false, queueArgs(properties, baseQueueName)); + declareQueue(baseQueueName, queue); + autoBindDLQ(baseQueueName, baseQueueName, properties); + org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue).to(exchange).with(name); + declareBinding(baseQueueName, binding); + } + else { + // if the stream is partitioned, create one queue for each target partition for the default group + for (int i = 0; i < properties.getNextModuleCount(); i++) { + String partitionSuffix = "-" + i; + String partitionQueueName = baseQueueName + partitionSuffix; + Queue queue = new Queue(partitionQueueName, true, false, false, + queueArgs(properties, partitionQueueName)); + declareQueue(queue.getName(), queue); + autoBindDLQ(baseQueueName, baseQueueName + partitionSuffix, properties); + declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange).with(name + partitionSuffix)); + } } } configureOutboundHandler(endpoint, properties); @@ -907,6 +924,11 @@ public class RabbitMessageChannelBinder extends AbstractBinder { */ private static final String REPUBLISH_TO_DLQ = "republishToDLQ"; + /** + * Durable pub/sub consumer. + */ + public static final String DURABLE = "durableSubscription"; + public RabbitPropertiesAccessor(Properties properties) { super(properties); } @@ -967,6 +989,10 @@ public class RabbitMessageChannelBinder extends AbstractBinder { return getProperty(REPUBLISH_TO_DLQ, defaultValue); } + public boolean isDurable(boolean defaultValue) { + return getProperty(DURABLE, defaultValue); + } + } } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitBinderConfigurationProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitBinderConfigurationProperties.java index 611cdf1bf..d21236442 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitBinderConfigurationProperties.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitBinderConfigurationProperties.java @@ -72,7 +72,7 @@ class RabbitBinderConfigurationProperties { private int compressionLevel; - private boolean durableSubscription; + private boolean durableSubscription = true; public AcknowledgeMode getAcknowledgeMode() { return acknowledgeMode; diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java index 07d697b34..14dff362c 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java @@ -78,6 +78,7 @@ public class RabbitMessageChannelBinderConfiguration { binder.setUsername(springRabbitMQProperties.getUsername()); binder.setUseSSL(springRabbitMQProperties.isUseSSL()); binder.setVhost(springRabbitMQProperties.getVhost()); + binder.setDefaultDurableSubscription(rabbitBinderConfigurationProperties.isDurableSubscription()); return binder; } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index 1cb06b8f2..b971c8214 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -340,7 +340,6 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { properties.put("maxAttempts", "1"); // disable retry properties.put("requeue", "false"); properties.put("partitionIndex", "0"); - properties.put("durableSubscription","true"); DirectChannel input0 = new DirectChannel(); input0.setBeanName("test.input0DLQ"); Binding input0Binding = binder.bindConsumer("partDLQ.0", "dlqPartGrp", input0, properties); @@ -424,6 +423,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { properties.put("prefix", "bindertest."); properties.put("autoBindDLQ", "true"); + properties.put("requiredGroups", "dlqPartGrp"); properties.put("partitionKeyExtractorClass", "org.springframework.cloud.stream.binder.PartitionTestSupport"); properties.put("partitionSelectorClass", "org.springframework.cloud.stream.binder.PartitionTestSupport"); properties.put(BinderPropertyKeys.NEXT_MODULE_COUNT, "2"); @@ -437,7 +437,6 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { properties.put("maxAttempts", "1"); // disable retry properties.put("requeue", "false"); properties.put("partitionIndex", "0"); - properties.put(BinderPropertyKeys.DURABLE,"true"); DirectChannel input0 = new DirectChannel(); input0.setBeanName("test.input0DLQ"); Binding input0Binding = binder.bindConsumer("partDLQ.1", "dlqPartGrp", input0, properties); @@ -563,6 +562,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { properties.put("batchBufferLimit", "100000"); properties.put("batchTimeout", "30000"); properties.put("compress", "true"); + properties.put("requiredGroups", "default"); DirectChannel output = new DirectChannel(); output.setBeanName("batchingProducer"); @@ -657,6 +657,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { MessageChannel outputChannel = new DirectChannel(); Binding pubSubProducerBinding = binder.bindProducer("latePubSub", outputChannel, properties); QueueChannel pubSubInputChannel = new QueueChannel(); + properties.setProperty("durableSubscription", "false"); Binding nonDurableConsumerBinding = binder.bindConsumer("latePubSub", "lategroup", pubSubInputChannel, properties); QueueChannel durablePubSubInputChannel = new QueueChannel(); properties.setProperty("durableSubscription", "true"); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java index b1894bc9c..bd54d77d2 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java @@ -55,6 +55,7 @@ import org.springframework.retry.RetryCallback; import org.springframework.retry.RetryContext; import org.springframework.retry.support.RetryTemplate; import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; /** @@ -85,7 +86,6 @@ public class RedisMessageChannelBinder extends AbstractBinder { .addAll(CONSUMER_RETRY_PROPERTIES) .add(BinderPropertyKeys.CONCURRENCY) .add(BinderPropertyKeys.PARTITION_INDEX) - .add(BinderPropertyKeys.DURABLE) .build(); /** @@ -94,6 +94,7 @@ public class RedisMessageChannelBinder extends AbstractBinder { private static final Set SUPPORTED_PRODUCER_PROPERTIES = new SetBuilder() .addAll(PRODUCER_PARTITIONING_PROPERTIES) .addAll(PRODUCER_STANDARD_PROPERTIES) + .add(BinderPropertyKeys.REQUIRED_GROUPS) .build(); private final RedisConnectionFactory connectionFactory; @@ -286,6 +287,12 @@ public class RedisMessageChannelBinder extends AbstractBinder { consumer.setBeanName("outbound." + name); consumer.afterPropertiesSet(); DefaultBinding producerBinding = new DefaultBinding<>(name, null, moduleOutputChannel, consumer, properties); + String[] requiredGroups = properties.getRequiredGroups(defaultRequiredGroups); + if (!ObjectUtils.isEmpty(requiredGroups)) { + for (String group : requiredGroups) { + this.redisOperations.boundZSetOps(CONSUMER_GROUPS_KEY_PREFIX + name).incrementScore(group, 1); + } + } consumer.start(); return producerBinding; } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java index d49c65255..0e1b372a6 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java @@ -105,6 +105,64 @@ abstract public class PartitionCapableBinderTests extends BrokerBinderTests { binding2.unbind(); } + @Test + public void testOneRequiredGroup() throws Exception { + Binder binder = getBinder(); + DirectChannel output = new DirectChannel(); + Properties properties = new Properties(); + + String testDestination = "testDestination" + UUID.randomUUID().toString().replace("-", ""); + + properties.put("requiredGroups", "test1"); + Binding producerBinding = binder.bindProducer(testDestination, output, properties); + + String testPayload = "foo-" + UUID.randomUUID().toString(); + output.send(new GenericMessage<>(testPayload.getBytes())); + + properties.clear(); + QueueChannel inbound1 = new QueueChannel(); + Binding consumerBinding = binder.bindConsumer(testDestination, "test1", inbound1, properties); + + Message receivedMessage1 = receive(inbound1); + assertThat(receivedMessage1, not(nullValue())); + assertThat(new String((byte[]) receivedMessage1.getPayload()), equalTo(testPayload)); + + producerBinding.unbind(); + consumerBinding.unbind(); + } + + @Test + public void testTwoRequiredGroups() throws Exception { + Binder binder = getBinder(); + DirectChannel output = new DirectChannel(); + Properties properties = new Properties(); + + String testDestination = "testDestination" + UUID.randomUUID().toString().replace("-", ""); + + properties.put("requiredGroups", "test1,test2"); + Binding producerBinding = binder.bindProducer(testDestination, output, properties); + + String testPayload = "foo-" + UUID.randomUUID().toString(); + output.send(new GenericMessage<>(testPayload.getBytes())); + + properties.clear(); + QueueChannel inbound1 = new QueueChannel(); + Binding consumerBinding1 = binder.bindConsumer(testDestination, "test1", inbound1, properties); + QueueChannel inbound2 = new QueueChannel(); + Binding consumerBinding2 = binder.bindConsumer(testDestination, "test2", inbound2, properties); + + Message receivedMessage1 = receive(inbound1); + assertThat(receivedMessage1, not(nullValue())); + assertThat(new String((byte[]) receivedMessage1.getPayload()), equalTo(testPayload)); + Message receivedMessage2 = receive(inbound2); + assertThat(receivedMessage2, not(nullValue())); + assertThat(new String((byte[]) receivedMessage2.getPayload()), equalTo(testPayload)); + + consumerBinding1.unbind(); + consumerBinding2.unbind(); + producerBinding.unbind(); + } + @Test public void testBadProperties() throws Exception { Binder binder = getBinder(); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java index 350fcdb7a..c45af9551 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java @@ -169,12 +169,12 @@ public abstract class AbstractBinder implements ApplicationContextAware, Init protected volatile long defaultBatchTimeout = DEFAULT_BATCH_TIMEOUT; + protected volatile String[] defaultRequiredGroups = new String[] {}; + // compression protected volatile boolean defaultCompress = false; - protected volatile boolean defaultDurableSubscription = false; - // Payload type cache private volatile Map> payloadTypeCache = new ConcurrentHashMap<>(); @@ -318,13 +318,6 @@ public abstract class AbstractBinder implements ApplicationContextAware, Init this.defaultCompress = defaultCompress; } - /** - * Set whether subscriptions to taps/topics are durable. - * @param defaultDurableSubscription true for durable (default false). - */ - public void setDefaultDurableSubscription(boolean defaultDurableSubscription) { - this.defaultDurableSubscription = defaultDurableSubscription; - } @Override public void afterPropertiesSet() throws Exception { @@ -338,8 +331,6 @@ public abstract class AbstractBinder implements ApplicationContextAware, Init public final Binding bindConsumer(String name, String group, T target, Properties properties) { DefaultBindingPropertiesAccessor accessor = new DefaultBindingPropertiesAccessor(properties); if (StringUtils.isEmpty(group)) { - Assert.isTrue(!accessor.getProperty(BinderPropertyKeys.DURABLE, defaultDurableSubscription), - "A consumer group is required for a durable subscription"); Assert.isTrue(accessor.getPartitionIndex() < 0, "A consumer group is required for a partitioned subscription"); } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderPropertyKeys.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderPropertyKeys.java index 9cb753b98..38cd19b59 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderPropertyKeys.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderPropertyKeys.java @@ -126,14 +126,15 @@ public abstract class BinderPropertyKeys { */ public static final String COMPRESS = "compress"; - /** - * Durable pub/sub consumer. - */ - public static final String DURABLE = "durableSubscription"; - /** * Minimum partition count, if the transport supports partitioning natively (e.g. Kafka) */ public static final String MIN_PARTITION_COUNT = "minPartitionCount"; + /** + * Required groups. The binder will ensure that consumers from these groups that bind after + * the producer will be able to receive messages produced in the mean time. + */ + public static final String REQUIRED_GROUPS = "requiredGroups"; + } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBindingPropertiesAccessor.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBindingPropertiesAccessor.java index 3df26354b..04001dadd 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBindingPropertiesAccessor.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBindingPropertiesAccessor.java @@ -337,14 +337,17 @@ public class DefaultBindingPropertiesAccessor { } /** - * If true, subscriptions to taps/topics will be durable. - * @param defaultValue the default value. - * @return the property or default value. + * A list of groups for which the binder will ensure message delivery, even if their consumers bind + * after the producer. This is a producer-property only. + * @param defaultValue the default value + * @return the property, parsed as a comma-separated list of values */ - public boolean isDurable(boolean defaultValue) { - return getProperty(BinderPropertyKeys.DURABLE, defaultValue); + public String[] getRequiredGroups(String[] defaultValue) { + String requiredGroupsValue = getProperty(BinderPropertyKeys.REQUIRED_GROUPS, ""); + return StringUtils.commaDelimitedListToStringArray(requiredGroupsValue); } + // Utility methods /** diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java index 4efb203ee..757036f16 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java @@ -16,6 +16,8 @@ package org.springframework.cloud.stream.config; +import org.springframework.util.StringUtils; + import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.annotation.JsonInclude.Include; @@ -55,6 +57,8 @@ public class BindingProperties { // Outbound properties + private String requiredGroups; + // Partition properties private String partitionKeyExpression; @@ -80,13 +84,10 @@ public class BindingProperties { private Integer batchTimeout; - // Inbound properties private Integer concurrency; - private Boolean durableSubscription; - // Partition properties private String partitionIndex; @@ -245,12 +246,12 @@ public class BindingProperties { this.partitioned = partitioned; } - public Boolean isDurableSubscription() { - return this.durableSubscription; + public String getRequiredGroups() { + return requiredGroups; } - public void setDurableSubscription(Boolean durableSubscription) { - this.durableSubscription = durableSubscription; + public void setRequiredGroups(String requiredGroups) { + this.requiredGroups = requiredGroups; } public String toString() { @@ -325,8 +326,8 @@ public class BindingProperties { sb.append("concurrency=" + this.concurrency); sb.append(COMMA); } - if (this.durableSubscription != null) { - sb.append("durableSubscription=" + this.durableSubscription); + if (!StringUtils.isEmpty(requiredGroups)) { + sb.append("requiredGroups=" + requiredGroups); sb.append(COMMA); } sb.deleteCharAt(sb.lastIndexOf(COMMA)); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java index 1e95a504b..cbf208270 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java @@ -122,10 +122,6 @@ public class ChannelBindingServiceProperties { channelConsumerProperties.setProperty(BinderPropertyKeys.CONCURRENCY, Integer.toString(bindingProperties.getConcurrency())); } - if (bindingProperties.isDurableSubscription() != null) { - channelConsumerProperties.setProperty(BinderPropertyKeys.DURABLE, - Boolean.toString(bindingProperties.isDurableSubscription())); - } updateConsumerPartitionProperties(inputChannelName, channelConsumerProperties); } return channelConsumerProperties; @@ -139,8 +135,15 @@ public class ChannelBindingServiceProperties { */ public Properties getProducerProperties(String outputChannelName) { Properties channelProducerProperties = new Properties(); - updateBatchProperties(outputChannelName, channelProducerProperties); - updateProducerPartitionProperties(outputChannelName, channelProducerProperties); + BindingProperties bindingProperties = this.bindings.get(outputChannelName); + if (bindingProperties != null) { + updateBatchProperties(bindingProperties, channelProducerProperties); + updateProducerPartitionProperties(bindingProperties, channelProducerProperties); + if (StringUtils.hasText(bindingProperties.getRequiredGroups())) { + channelProducerProperties.setProperty(BinderPropertyKeys.REQUIRED_GROUPS, + bindingProperties.getRequiredGroups()); + } + } return channelProducerProperties; } @@ -149,62 +152,55 @@ public class ChannelBindingServiceProperties { return bindingProperties != null && bindingProperties.isPartitioned(); } - private boolean isPartitionedProducer(String channelName) { - BindingProperties bindingProperties = bindings.get(channelName); - return (bindingProperties != null && (StringUtils.hasText(bindingProperties.getPartitionKeyExpression()) - || StringUtils.hasText(bindingProperties.getPartitionKeyExtractorClass()))); + private boolean isPartitionedProducer(BindingProperties bindingProperties) { + return (StringUtils.hasText(bindingProperties.getPartitionKeyExpression()) + || StringUtils.hasText(bindingProperties.getPartitionKeyExtractorClass())); } - private void updateBatchProperties(String outputChannelName, Properties producerProperties) { - BindingProperties bindingProperties = this.bindings.get(outputChannelName); - if (bindingProperties != null) { - if (bindingProperties.isBatchingEnabled() != null) { - producerProperties.setProperty(BinderPropertyKeys.BATCHING_ENABLED, - String.valueOf(bindingProperties.isBatchingEnabled())); - } - if (bindingProperties.getBatchSize() != null) { - producerProperties.setProperty(BinderPropertyKeys.BATCH_SIZE, - String.valueOf(bindingProperties.getBatchSize())); - } - if (bindingProperties.getBatchBufferLimit() != null) { - producerProperties.setProperty(BinderPropertyKeys.BATCH_BUFFER_LIMIT, - String.valueOf(bindingProperties.getBatchBufferLimit())); - } - if (bindingProperties.getBatchTimeout() != null) { - producerProperties.setProperty(BinderPropertyKeys.BATCH_TIMEOUT, - String.valueOf(bindingProperties.getBatchTimeout())); - } + private void updateBatchProperties(BindingProperties bindingProperties, Properties producerProperties) { + if (bindingProperties.isBatchingEnabled() != null) { + producerProperties.setProperty(BinderPropertyKeys.BATCHING_ENABLED, + String.valueOf(bindingProperties.isBatchingEnabled())); + } + if (bindingProperties.getBatchSize() != null) { + producerProperties.setProperty(BinderPropertyKeys.BATCH_SIZE, + String.valueOf(bindingProperties.getBatchSize())); + } + if (bindingProperties.getBatchBufferLimit() != null) { + producerProperties.setProperty(BinderPropertyKeys.BATCH_BUFFER_LIMIT, + String.valueOf(bindingProperties.getBatchBufferLimit())); + } + if (bindingProperties.getBatchTimeout() != null) { + producerProperties.setProperty(BinderPropertyKeys.BATCH_TIMEOUT, + String.valueOf(bindingProperties.getBatchTimeout())); } } - private void updateProducerPartitionProperties(String outputChannelName, Properties producerProperties) { - BindingProperties bindingProperties = this.bindings.get(outputChannelName); - if (bindingProperties != null) { - if (isPartitionedProducer(outputChannelName)) { - if (bindingProperties.getPartitionKeyExpression() != null) { - producerProperties.setProperty(BinderPropertyKeys.PARTITION_KEY_EXPRESSION, - bindingProperties.getPartitionKeyExpression()); - } - if (bindingProperties.getPartitionKeyExtractorClass() != null) { - producerProperties.setProperty(BinderPropertyKeys.PARTITION_KEY_EXTRACTOR_CLASS, - bindingProperties.getPartitionKeyExtractorClass()); - } - if (bindingProperties.getPartitionSelectorClass() != null) { - producerProperties.setProperty(BinderPropertyKeys.PARTITION_SELECTOR_CLASS, - bindingProperties.getPartitionSelectorClass()); - } - if (bindingProperties.getPartitionSelectorExpression() != null) { - producerProperties.setProperty(BinderPropertyKeys.PARTITION_SELECTOR_EXPRESSION, - bindingProperties.getPartitionSelectorExpression()); - } - if (bindingProperties.getPartitionCount() != null) { - producerProperties.setProperty(BinderPropertyKeys.NEXT_MODULE_COUNT, - Integer.toString(bindingProperties.getPartitionCount())); - } - if (bindingProperties.getNextModuleConcurrency() != null) { - producerProperties.setProperty(BinderPropertyKeys.NEXT_MODULE_CONCURRENCY, - Integer.toString(bindingProperties.getNextModuleConcurrency())); - } + private void updateProducerPartitionProperties(BindingProperties bindingProperties, Properties producerProperties) { + if (isPartitionedProducer(bindingProperties)) { + if (bindingProperties.getPartitionKeyExpression() != null) { + producerProperties.setProperty(BinderPropertyKeys.PARTITION_KEY_EXPRESSION, + bindingProperties.getPartitionKeyExpression()); + } + if (bindingProperties.getPartitionKeyExtractorClass() != null) { + producerProperties.setProperty(BinderPropertyKeys.PARTITION_KEY_EXTRACTOR_CLASS, + bindingProperties.getPartitionKeyExtractorClass()); + } + if (bindingProperties.getPartitionSelectorClass() != null) { + producerProperties.setProperty(BinderPropertyKeys.PARTITION_SELECTOR_CLASS, + bindingProperties.getPartitionSelectorClass()); + } + if (bindingProperties.getPartitionSelectorExpression() != null) { + producerProperties.setProperty(BinderPropertyKeys.PARTITION_SELECTOR_EXPRESSION, + bindingProperties.getPartitionSelectorExpression()); + } + if (bindingProperties.getPartitionCount() != null) { + producerProperties.setProperty(BinderPropertyKeys.NEXT_MODULE_COUNT, + Integer.toString(bindingProperties.getPartitionCount())); + } + if (bindingProperties.getNextModuleConcurrency() != null) { + producerProperties.setProperty(BinderPropertyKeys.NEXT_MODULE_CONCURRENCY, + Integer.toString(bindingProperties.getNextModuleConcurrency())); } } }