diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java index 45cccc874..6f0d26d2b 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java +++ b/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 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; } diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java index f1aaaec03..d885dd3f6 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java +++ b/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 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(), diff --git a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc index 668908df9..ccfa9315c 100644 --- a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc +++ b/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 <> 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` to receive these events See <> 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 diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index fb1e0420b..8f94bbe84 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/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 allPartitions = getPartitionInfo(destination, extendedConsumerProperties, - consumerFactory, partitionCount); + boolean usingPatterns = extendedConsumerProperties.getExtension().isDestinationIsPattern(); + Collection allPartitions = usingPatterns ? Collections.emptyList() + : getPartitionInfo(destination, extendedConsumerProperties, consumerFactory, partitionCount); Collection 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() diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 8bc3f1ba9..78fc0be46 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/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 consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setDestinationIsPattern(true); + DirectChannel moduleInputChannel = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicReference topic = new AtomicReference<>(); + moduleInputChannel.subscribe(m -> { + topic.set(m.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC, String.class)); + latch.countDown(); + }); + Binding 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;