diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java index 94ce0f573..2ee9bd67b 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java @@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binder.kafka.streams; import java.util.Arrays; import java.util.Map; +import java.util.regex.Pattern; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -75,17 +76,21 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application private static final Log LOG = LogFactory.getLog(AbstractKafkaStreamsBinderProcessor.class); private final KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue; + private final BindingServiceProperties bindingServiceProperties; + private final KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties; + private final CleanupConfig cleanupConfig; + private final KeyValueSerdeResolver keyValueSerdeResolver; protected ConfigurableApplicationContext applicationContext; public AbstractKafkaStreamsBinderProcessor(BindingServiceProperties bindingServiceProperties, - KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue, - KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties, - KeyValueSerdeResolver keyValueSerdeResolver, CleanupConfig cleanupConfig) { + KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue, + KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties, + KeyValueSerdeResolver keyValueSerdeResolver, CleanupConfig cleanupConfig) { this.bindingServiceProperties = bindingServiceProperties; this.kafkaStreamsBindingInformationCatalogue = kafkaStreamsBindingInformationCatalogue; this.kafkaStreamsExtendedBindingProperties = kafkaStreamsExtendedBindingProperties; @@ -125,9 +130,9 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application @SuppressWarnings("unchecked") protected void handleKTableGlobalKTableInputs(Object[] arguments, int index, String input, Class parameterType, Object targetBean, - StreamsBuilderFactoryBean streamsBuilderFactoryBean, StreamsBuilder streamsBuilder, - KafkaStreamsConsumerProperties extendedConsumerProperties, - Serde keySerde, Serde valueSerde, Topology.AutoOffsetReset autoOffsetReset) { + StreamsBuilderFactoryBean streamsBuilderFactoryBean, StreamsBuilder streamsBuilder, + KafkaStreamsConsumerProperties extendedConsumerProperties, + Serde keySerde, Serde valueSerde, Topology.AutoOffsetReset autoOffsetReset) { if (parameterType.isAssignableFrom(KTable.class)) { String materializedAs = extendedConsumerProperties.getMaterializedAs(); String bindingDestination = this.bindingServiceProperties.getBindingDestination(input); @@ -154,7 +159,7 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application } } - @SuppressWarnings({"unchecked"}) + @SuppressWarnings({ "unchecked" }) protected StreamsBuilderFactoryBean buildStreamsBuilderAndRetrieveConfig(String beanNamePostPrefix, ApplicationContext applicationContext, String inboundName, KafkaStreamsBinderConfigurationProperties kafkaStreamsBinderConfigurationProperties, @@ -329,16 +334,24 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application } protected KStream getKStream(String inboundName, BindingProperties bindingProperties, KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties, - StreamsBuilder streamsBuilder, Serde keySerde, Serde valueSerde, Topology.AutoOffsetReset autoOffsetReset, boolean firstBuild) { + StreamsBuilder streamsBuilder, Serde keySerde, Serde valueSerde, Topology.AutoOffsetReset autoOffsetReset, boolean firstBuild) { if (firstBuild) { addStateStoreBeans(streamsBuilder); } - String[] bindingTargets = StringUtils.commaDelimitedListToStringArray( - this.bindingServiceProperties.getBindingDestination(inboundName)); - final Consumed consumed = getConsumed(kafkaStreamsConsumerProperties, keySerde, valueSerde, autoOffsetReset); - KStream stream = streamsBuilder.stream(Arrays.asList(bindingTargets), - consumed); + KStream stream; + if (this.kafkaStreamsExtendedBindingProperties + .getExtendedConsumerProperties(inboundName).isDestinationIsPattern()) { + final Pattern pattern = Pattern.compile(this.bindingServiceProperties.getBindingDestination(inboundName)); + stream = streamsBuilder.stream(pattern); + } + else { + String[] bindingTargets = StringUtils.commaDelimitedListToStringArray( + this.bindingServiceProperties.getBindingDestination(inboundName)); + final Consumed consumed = getConsumed(kafkaStreamsConsumerProperties, keySerde, valueSerde, autoOffsetReset); + stream = streamsBuilder.stream(Arrays.asList(bindingTargets), + consumed); + } final boolean nativeDecoding = this.bindingServiceProperties .getConsumerProperties(inboundName).isUseNativeDecoding(); if (nativeDecoding) { @@ -390,7 +403,7 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application } private KTable materializedAs(StreamsBuilder streamsBuilder, String destination, String storeName, - Serde k, Serde v, Topology.AutoOffsetReset autoOffsetReset, KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties) { + Serde k, Serde v, Topology.AutoOffsetReset autoOffsetReset, KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties) { final Consumed consumed = getConsumed(kafkaStreamsConsumerProperties, k, v, autoOffsetReset); return streamsBuilder.table(this.bindingServiceProperties.getBindingDestination(destination), @@ -414,9 +427,9 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application } private GlobalKTable getGlobalKTable(KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties, - StreamsBuilder streamsBuilder, - Serde keySerde, Serde valueSerde, String materializedAs, - String bindingDestination, Topology.AutoOffsetReset autoOffsetReset) { + StreamsBuilder streamsBuilder, + Serde keySerde, Serde valueSerde, String materializedAs, + String bindingDestination, Topology.AutoOffsetReset autoOffsetReset) { final Consumed consumed = getConsumed(kafkaStreamsConsumerProperties, keySerde, valueSerde, autoOffsetReset); return materializedAs != null ? materializedAsGlobalKTable(streamsBuilder, bindingDestination, @@ -426,9 +439,9 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application } private KTable getKTable(KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties, - StreamsBuilder streamsBuilder, Serde keySerde, - Serde valueSerde, String materializedAs, String bindingDestination, - Topology.AutoOffsetReset autoOffsetReset) { + StreamsBuilder streamsBuilder, Serde keySerde, + Serde valueSerde, String materializedAs, String bindingDestination, + Topology.AutoOffsetReset autoOffsetReset) { final Consumed consumed = getConsumed(kafkaStreamsConsumerProperties, keySerde, valueSerde, autoOffsetReset); return materializedAs != null ? materializedAs(streamsBuilder, bindingDestination, materializedAs, @@ -438,7 +451,7 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application } private Consumed getConsumed(KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties, - Serde keySerde, Serde valueSerde, Topology.AutoOffsetReset autoOffsetReset) { + Serde keySerde, Serde valueSerde, Topology.AutoOffsetReset autoOffsetReset) { TimestampExtractor timestampExtractor = null; if (!StringUtils.isEmpty(kafkaStreamsConsumerProperties.getTimestampExtractorBeanName())) { timestampExtractor = applicationContext.getBean(kafkaStreamsConsumerProperties.getTimestampExtractorBeanName(), diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderDestinationIsPatternTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderDestinationIsPatternTests.java new file mode 100644 index 000000000..424b87cd8 --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderDestinationIsPatternTests.java @@ -0,0 +1,112 @@ +/* + * Copyright 2019-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kafka.streams.integration; + +import java.util.Map; +import java.util.function.Function; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.streams.kstream.KStream; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.utils.KafkaTestUtils; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Michael Stoettinger + */ +public class KafkaStreamsBinderDestinationIsPatternTests { + + @ClassRule + public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, + "in.1", "in.2", "out"); + + private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule + .getEmbeddedKafka(); + + private static org.apache.kafka.clients.consumer.Consumer consumer; + + @BeforeClass + public static void setUp() { + Map consumerProps = KafkaTestUtils.consumerProps("group", "true", + embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + consumer = cf.createConsumer(); + embeddedKafka.consumeFromEmbeddedTopics(consumer, "out"); + } + + @AfterClass + public static void tearDown() { + consumer.close(); + } + + @Test + public void test() { + SpringApplication app = new SpringApplication(ConsumingApplication.class); + app.setWebApplicationType(WebApplicationType.NONE); + ConfigurableApplicationContext context = app.run("--server.port=0", + "--spring.cloud.stream.bindings.process-out-0.destination=out", + "--spring.cloud.stream.bindings.process-in-0.destination=in.*", + "--spring.cloud.stream.bindings.process-in-0.consumer.use-native-decoding=false", + "--spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.destinationIsPattern=true", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + + embeddedKafka.getBrokersAsString()); + try { + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( + senderProps); + KafkaTemplate template = new KafkaTemplate<>(producerFactory, true); + + // send message to both topics that fit the pattern + template.send("in.1", "foo1"); + assertThat(KafkaTestUtils.getSingleRecord(consumer, "out").value()) + .isEqualTo("foo1"); + template.send("in.2", "foo2"); + assertThat(KafkaTestUtils.getSingleRecord(consumer, "out").value()) + .isEqualTo("foo2"); + } + finally { + context.close(); + } + } + + @EnableAutoConfiguration + public static class ConsumingApplication { + + @Bean + public Function, KStream> process() { + return input -> input; + } + + } + +}