GH-339: Support topic patterns

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/339

- support regex patterns for consumer destinations to consume from multiple topics
- `enableDlq` is not available in this mode

As well as the test case, tested with a boot app...

```
spring.cloud.stream.bindings.input.destination=kbgh339.*
spring.cloud.stream.kafka.bindings.input.consumer.destination-is-pattern=true
```

and

```
2018-04-30 15:12:49.718  : partitions assigned: [kbgh339a-0]
2018-04-30 15:17:46.585  : partitions revoked: [kbgh339a-0]
2018-04-30 15:17:46.655  : partitions assigned: [kbgh339a-0, kbgh339b-0]
```

after adding a new topic matching the pattern.

Doc polishing.

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java

@@ -81,6 +81,8 @@ public class KafkaConsumerProperties {
 
 	private long idleEventInterval = 30_000;
 
+	private boolean destinationIsPattern;
+
 	private Map<String, String> configuration = new HashMap<>();
 
 	private KafkaAdminProperties admin = new KafkaAdminProperties();
@@ -216,6 +218,14 @@ public class KafkaConsumerProperties {
 		this.idleEventInterval = idleEventInterval;
 	}
 
+	public boolean isDestinationIsPattern() {
+		return this.destinationIsPattern;
+	}
+
+	public void setDestinationIsPattern(boolean destinationIsPattern) {
+		this.destinationIsPattern = destinationIsPattern;
+	}
+
 	public KafkaAdminProperties getAdmin() {
 		return this.admin;
 	}

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java

@@ -152,6 +152,15 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 	public ConsumerDestination provisionConsumerDestination(final String name, final String group,
 			ExtendedConsumerProperties<KafkaConsumerProperties> properties) {
 
+		if (properties.getExtension().isDestinationIsPattern()) {
+			Assert.isTrue(!properties.getExtension().isEnableDlq(),
+					"enableDLQ is not allowed when listening to topic patterns");
+			if (this.logger.isDebugEnabled()) {
+				this.logger.debug("Listening to a topic pattern - " + name
+						+ " - no provisioning performed");
+			}
+			return new KafkaConsumerDestination(name);
+		}
 		KafkaTopicUtils.validateTopicName(name);
 		boolean anonymous = !StringUtils.hasText(group);
 		Assert.isTrue(!anonymous || !properties.getExtension().isEnableDlq(),

spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc

@@ -207,6 +207,7 @@ The DLQ topic name can be configurable by setting the `dlqName` property.
 This provides an alternative option to the more common Kafka replay scenario for the case when the number of errors is relatively small and replaying the entire original topic may be too cumbersome.
 See <<kafka-dlq-processing>> processing for more information.
 Starting with version 2.0, messages sent to the DLQ topic are enhanced with the following headers: `x-original-topic`, `x-exception-message`, and `x-exception-stacktrace` as `byte[]`.
+**Not allowed when `destinationIsPattern` is `true`.**
 +
 Default: `false`.
 configuration::
@@ -238,6 +239,13 @@ Use an `ApplicationListener<ListenerContainerIdleEvent>` to receive these events
 See <<pause-resume>> for a usage example.
 +
 Default: `30000`
+destinationIsPattern::
+When true, the destination is treated as a regular expression `Pattern` used to match topic names by the broker.
+When true, topics are not provisioned, and `enableDlq` is not allowed, because the binder does not know the topic names during the provisioning phase.
+Note, the time taken to detect new topics that match the pattern is controlled by the consumer property `metadata.max.age.ms`, which (at the time of writing) defaults to 300,000ms (5 minutes).
+This can be configured using the `configuration` property above.
++
+Default: `false`
 
 [[kafka-producer-properties]]
 === Kafka Producer Properties

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

@@ -22,6 +22,7 @@ import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
@@ -31,6 +32,7 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Predicate;
+import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
 import org.apache.kafka.clients.consumer.Consumer;
@@ -329,8 +331,9 @@ public class KafkaMessageChannelBinder extends
 		int partitionCount = extendedConsumerProperties.getInstanceCount()
 				* extendedConsumerProperties.getConcurrency();
 
-		Collection<PartitionInfo> allPartitions = getPartitionInfo(destination, extendedConsumerProperties,
-				consumerFactory, partitionCount);
+		boolean usingPatterns = extendedConsumerProperties.getExtension().isDestinationIsPattern();
+		Collection<PartitionInfo> allPartitions = usingPatterns ? Collections.emptyList()
+				: getPartitionInfo(destination, extendedConsumerProperties, consumerFactory, partitionCount);
 
 		Collection<PartitionInfo> listenedPartitions;
 
@@ -350,20 +353,25 @@ public class KafkaMessageChannelBinder extends
 				}
 			}
 		}
-		this.topicsInUse.put(destination.getName(), new TopicInformation(group, listenedPartitions));
+		String topics = destination.getName();
+		this.topicsInUse.put(topics, new TopicInformation(group, listenedPartitions));
 
-		Assert.isTrue(!CollectionUtils.isEmpty(listenedPartitions), "A list of partitions must be provided");
+		Assert.isTrue(usingPatterns
+				|| !CollectionUtils.isEmpty(listenedPartitions), "A list of partitions must be provided");
 		final TopicPartitionInitialOffset[] topicPartitionInitialOffsets = getTopicPartitionInitialOffsets(
 				listenedPartitions);
 		final ContainerProperties containerProperties = anonymous
 				|| extendedConsumerProperties.getExtension().isAutoRebalanceEnabled()
-						? new ContainerProperties(destination.getName())
+						? usingPatterns
+								? new ContainerProperties(Pattern.compile(topics))
+								: new ContainerProperties(topics)
 						: new ContainerProperties(topicPartitionInitialOffsets);
 		if (this.transactionManager != null) {
 			containerProperties.setTransactionManager(this.transactionManager);
 		}
 		containerProperties.setIdleEventInterval(extendedConsumerProperties.getExtension().getIdleEventInterval());
-		int concurrency = Math.min(extendedConsumerProperties.getConcurrency(), listenedPartitions.size());
+		int concurrency = usingPatterns ? extendedConsumerProperties.getConcurrency()
+				: Math.min(extendedConsumerProperties.getConcurrency(), listenedPartitions.size());
 		resetOffsets(extendedConsumerProperties, consumerFactory, groupManagement, containerProperties);
 		@SuppressWarnings("rawtypes")
 		final ConcurrentMessageListenerContainer<?, ?> messageListenerContainer =
@@ -383,7 +391,7 @@ public class KafkaMessageChannelBinder extends
 		else if (getApplicationContext() != null) {
 			messageListenerContainer.setApplicationEventPublisher(getApplicationContext());
 		}
-		messageListenerContainer.setBeanName(destination.getName() + ".container");
+		messageListenerContainer.setBeanName(topics + ".container");
 		// end of these won't be needed...
 		if (!extendedConsumerProperties.getExtension().isAutoCommitOffset()) {
 			messageListenerContainer.getContainerProperties()

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java

@@ -1778,8 +1778,6 @@ public class KafkaBinderTests extends
 	@SuppressWarnings("unchecked")
 	public void testDefaultConsumerStartsAtEarliest() throws Exception {
 		Binder binder = getBinder(createConfigurationProperties());
-		GenericApplicationContext context = new GenericApplicationContext();
-		context.refresh();
 
 		BindingProperties producerBindingProperties = createProducerBindingProperties(createProducerProperties());
 		DirectChannel output = createBindableChannel("output", producerBindingProperties);
@@ -2510,6 +2508,35 @@ public class KafkaBinderTests extends
 		consumer.close();
 	}
 
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@Test
+	public void testTopicPatterns() throws Exception {
+		try (AdminClient admin = AdminClient.create(Collections.singletonMap(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
+				embeddedKafka.getBrokersAsString()))) {
+			admin.createTopics(Collections.singletonList(new NewTopic("topicPatterns.1", 1, (short) 1))).all().get();
+			Binder binder = getBinder();
+			ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
+			consumerProperties.getExtension().setDestinationIsPattern(true);
+			DirectChannel moduleInputChannel = createBindableChannel("input", createConsumerBindingProperties(consumerProperties));
+			final CountDownLatch latch = new CountDownLatch(1);
+			final AtomicReference<String> topic = new AtomicReference<>();
+			moduleInputChannel.subscribe(m -> {
+				topic.set(m.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC, String.class));
+				latch.countDown();
+			});
+			Binding<MessageChannel> consumerBinding = binder.bindConsumer("topicPatterns\\..*",
+					"testTopicPatterns", moduleInputChannel, consumerProperties);
+			DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory(
+					KafkaTestUtils.producerProps(embeddedKafka));
+			KafkaTemplate template = new KafkaTemplate(pf);
+			template.send("topicPatterns.1", "foo");
+			assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+			assertThat(topic.get()).isEqualTo("topicPatterns.1");
+			consumerBinding.unbind();
+			pf.destroy();
+		}
+	}
+
 	private final class FailingInvocationCountingMessageHandler implements MessageHandler {
 
 		private int invocationCount;