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

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<MessageChannel> {
 
 	private static final Set<Object> KAFKA_PRODUCER_PROPERTIES = new SetBuilder()
 			.add(BinderPropertyKeys.MIN_PARTITION_COUNT)
+			.add(BinderPropertyKeys.REQUIRED_GROUPS)
 			.build();
 
 	/**
@@ -249,7 +250,7 @@ public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel> {
 
 	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<MessageChannel> {
 		// 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);
 	}
 

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<byte[]> receivedMessage1 = (Message<byte[]>) 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<byte[]> receivedMessage2 = (Message<byte[]>) receive(input1);

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<MessageChannel> {
 	private static final String DEAD_LETTER_EXCHANGE = "DLX";
 
 	private static final Set<Object> RABBIT_CONSUMER_PROPERTIES = new HashSet<Object>(Arrays.asList(new String[] {
-
 		BinderPropertyKeys.MAX_CONCURRENCY,
 		RabbitPropertiesAccessor.ACK_MODE,
 		RabbitPropertiesAccessor.PREFETCH,
@@ -144,7 +143,8 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel> {
 		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<MessageChannel> {
 	 */
 	private static final Set<Object> 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<MessageChannel> {
 			.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<MessageChannel> {
 
 	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<MessageChannel> {
 		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<MessageChannel> {
 		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<MessageChannel> {
 		 */
 		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<MessageChannel> {
 			return getProperty(REPUBLISH_TO_DLQ, defaultValue);
 		}
 
+		public boolean isDurable(boolean defaultValue) {
+			return getProperty(DURABLE, defaultValue);
+		}
+
 	}
 
 }

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;

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;
 	}
 	

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<MessageChannel> 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<MessageChannel> 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<MessageChannel> pubSubProducerBinding = binder.bindProducer("latePubSub", outputChannel, properties);
 		QueueChannel pubSubInputChannel = new QueueChannel();
+		properties.setProperty("durableSubscription", "false");
 		Binding<MessageChannel> nonDurableConsumerBinding = binder.bindConsumer("latePubSub", "lategroup", pubSubInputChannel, properties);
 		QueueChannel durablePubSubInputChannel = new QueueChannel();
 		properties.setProperty("durableSubscription", "true");

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<MessageChannel> {
 			.addAll(CONSUMER_RETRY_PROPERTIES)
 			.add(BinderPropertyKeys.CONCURRENCY)
 			.add(BinderPropertyKeys.PARTITION_INDEX)
-			.add(BinderPropertyKeys.DURABLE)
 			.build();
 
 	/**
@@ -94,6 +94,7 @@ public class RedisMessageChannelBinder extends AbstractBinder<MessageChannel> {
 	private static final Set<Object> 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<MessageChannel> {
 		consumer.setBeanName("outbound." + name);
 		consumer.afterPropertiesSet();
 		DefaultBinding<MessageChannel> 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;
 	}

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<MessageChannel> binder = getBinder();
+		DirectChannel output = new DirectChannel();
+		Properties properties = new Properties();
+
+		String testDestination = "testDestination" + UUID.randomUUID().toString().replace("-", "");
+
+		properties.put("requiredGroups", "test1");
+		Binding<MessageChannel> 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<MessageChannel> 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<MessageChannel> binder = getBinder();
+		DirectChannel output = new DirectChannel();
+		Properties properties = new Properties();
+
+		String testDestination = "testDestination" + UUID.randomUUID().toString().replace("-", "");
+
+		properties.put("requiredGroups", "test1,test2");
+		Binding<MessageChannel> 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<MessageChannel> consumerBinding1 = binder.bindConsumer(testDestination, "test1", inbound1, properties);
+		QueueChannel inbound2 = new QueueChannel();
+		Binding<MessageChannel> 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<MessageChannel> binder = getBinder();

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java

@@ -169,12 +169,12 @@ public abstract class AbstractBinder<T> 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<String, Class<?>> payloadTypeCache = new ConcurrentHashMap<>();
 
@@ -318,13 +318,6 @@ public abstract class AbstractBinder<T> 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<T> implements ApplicationContextAware, Init
 	public final Binding<T> 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");
 		}

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";
+
 }

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
 
 	/**

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));

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()));
 			}
 		}
 	}