diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamApplicationSupportAutoConfiguration.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamApplicationSupportAutoConfiguration.java new file mode 100644 index 000000000..b16b7a9b4 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamApplicationSupportAutoConfiguration.java @@ -0,0 +1,40 @@ +/* + * Copyright 2017 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 + * + * http://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.kstream.config; + +import org.apache.kafka.streams.kstream.TimeWindows; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * @author Soby Chacko + */ +@Configuration +@EnableConfigurationProperties(KStreamApplicationSupportProperties.class) +public class KStreamApplicationSupportAutoConfiguration { + + @Bean + @ConditionalOnProperty("spring.cloud.stream.kstream.timeWindow.length") + public TimeWindows configuredTimeWindow(KStreamApplicationSupportProperties processorProperties) { + return processorProperties.getTimeWindow().getAdvanceBy() > 0 + ? TimeWindows.of(processorProperties.getTimeWindow().getLength()).advanceBy(processorProperties.getTimeWindow().getAdvanceBy()) + : TimeWindows.of(processorProperties.getTimeWindow().getLength()); + } +} diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamApplicationSupportProperties.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamApplicationSupportProperties.java new file mode 100644 index 000000000..031f01279 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamApplicationSupportProperties.java @@ -0,0 +1,64 @@ +/* + * Copyright 2017 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 + * + * http://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.kstream.config; + +import org.springframework.boot.context.properties.ConfigurationProperties; + +/** + * {@link ConfigurationProperties} that can be used by end user Kafka Stream applications. This class provides + * convenient ways to access the commonly used kafka stream properties from the user application. For example, windowing + * operations are common use cases in stream processing and one can provide window specific properties at runtime and use + * those properties in the applications using this class. + * + * @author Soby Chacko + */ +@ConfigurationProperties("spring.cloud.stream.kstream") +public class KStreamApplicationSupportProperties { + + private TimeWindow timeWindow; + + public TimeWindow getTimeWindow() { + return timeWindow; + } + + public void setTimeWindow(TimeWindow timeWindow) { + this.timeWindow = timeWindow; + } + + public static class TimeWindow { + + private int length; + + private int advanceBy; + + public int getLength() { + return length; + } + + public void setLength(int length) { + this.length = length; + } + + public int getAdvanceBy() { + return advanceBy; + } + + public void setAdvanceBy(int advanceBy) { + this.advanceBy = advanceBy; + } + } +} diff --git a/spring-cloud-stream-binder-kstream/src/main/resources/META-INF/spring.factories b/spring-cloud-stream-binder-kstream/src/main/resources/META-INF/spring.factories index 7246dbc7c..adeb7d6a2 100644 --- a/spring-cloud-stream-binder-kstream/src/main/resources/META-INF/spring.factories +++ b/spring-cloud-stream-binder-kstream/src/main/resources/META-INF/spring.factories @@ -1,4 +1,5 @@ org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ - org.springframework.cloud.stream.binder.kstream.config.KStreamBinderSupportAutoConfiguration + org.springframework.cloud.stream.binder.kstream.config.KStreamBinderSupportAutoConfiguration,\ + org.springframework.cloud.stream.binder.kstream.config.KStreamApplicationSupportAutoConfiguration diff --git a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderWordCountIntegrationTests.java b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderWordCountIntegrationTests.java index a60a956bc..819078124 100644 --- a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderWordCountIntegrationTests.java +++ b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderWordCountIntegrationTests.java @@ -39,15 +39,14 @@ import org.junit.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.annotation.StreamListener; import org.springframework.cloud.stream.binder.kstream.annotations.KStreamProcessor; +import org.springframework.cloud.stream.binder.kstream.config.KStreamApplicationSupportProperties; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; -import org.springframework.kafka.core.KStreamBuilderFactoryBean; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.test.rule.KafkaEmbedded; import org.springframework.kafka.test.utils.KafkaTestUtils; @@ -96,6 +95,8 @@ public class KStreamBinderWordCountIntegrationTests { "--spring.cloud.stream.bindings.output.producer.headerMode=raw", "--spring.cloud.stream.bindings.output.producer.useNativeEncoding=true", "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", + "--spring.cloud.stream.kstream.timeWindow.length=5000", + "--spring.cloud.stream.kstream.timeWindow.advanceBy=0", "--spring.cloud.stream.kstream.binder.brokers=" + embeddedKafka.getBrokersAsString(), "--spring.cloud.stream.kstream.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); receiveAndValidate(context); @@ -114,14 +115,11 @@ public class KStreamBinderWordCountIntegrationTests { @EnableBinding(KStreamProcessor.class) @EnableAutoConfiguration - @EnableConfigurationProperties(WordCountProcessorProperties.class) + @EnableConfigurationProperties(KStreamApplicationSupportProperties.class) public static class WordCountProcessorApplication { @Autowired - private WordCountProcessorProperties processorProperties; - - @Autowired - private KStreamBuilderFactoryBean kafkaStreams; + private TimeWindows timeWindows; @StreamListener("input") @SendTo("output") @@ -143,7 +141,7 @@ public class KStreamBinderWordCountIntegrationTests { } }) .groupByKey(Serdes.String(), Serdes.String()) - .count(configuredTimeWindow(), processorProperties.getStoreName()) + .count(timeWindows, "WordCounts") .toStream() .map(new KeyValueMapper, Long, KeyValue>() { @@ -154,50 +152,6 @@ public class KStreamBinderWordCountIntegrationTests { }); } - /** - * Constructs a {@link TimeWindows} property. - * - * @return - */ - private TimeWindows configuredTimeWindow() { - return processorProperties.getAdvanceBy() > 0 - ? TimeWindows.of(processorProperties.getWindowLength()).advanceBy(processorProperties.getAdvanceBy()) - : TimeWindows.of(processorProperties.getWindowLength()); - } - } - - @ConfigurationProperties(prefix = "kstream.word.count") - static class WordCountProcessorProperties { - - private int windowLength = 5000; - - private int advanceBy = 0; - - private String storeName = "WordCounts"; - - int getWindowLength() { - return windowLength; - } - - public void setWindowLength(int windowLength) { - this.windowLength = windowLength; - } - - int getAdvanceBy() { - return advanceBy; - } - - public void setAdvanceBy(int advanceBy) { - this.advanceBy = advanceBy; - } - - String getStoreName() { - return storeName; - } - - public void setStoreName(String storeName) { - this.storeName = storeName; - } } static class WordCount {