diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/CompositeKafkaStreamsCustomizer.java b/spring-kafka/src/main/java/org/springframework/kafka/core/CompositeKafkaStreamsCustomizer.java new file mode 100644 index 00000000..c8e3bdff --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/CompositeKafkaStreamsCustomizer.java @@ -0,0 +1,53 @@ +/* + * Copyright 2018 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.kafka.core; + +import java.util.ArrayList; +import java.util.List; + +import org.apache.kafka.streams.KafkaStreams; + +/** + * Composite {@link KafkaStreamsCustomizer} customizes {@link KafkaStreams} by delegating + * to a list of provided {@link KafkaStreamsCustomizer}. + * + * @author Nurettin Yilmaz + * @author Artem Bilan + * + * @since 2.1.5 + */ +public class CompositeKafkaStreamsCustomizer implements KafkaStreamsCustomizer { + + private final List kafkaStreamsCustomizers = new ArrayList<>(); + + public CompositeKafkaStreamsCustomizer() { + } + + public CompositeKafkaStreamsCustomizer(List kafkaStreamsCustomizers) { + this.kafkaStreamsCustomizers.addAll(kafkaStreamsCustomizers); + } + + @Override + public void customize(KafkaStreams kafkaStreams) { + this.kafkaStreamsCustomizers.forEach(kafkaStreamsCustomizer -> kafkaStreamsCustomizer.customize(kafkaStreams)); + } + + public void addKafkaStreamsCustomizers(List kafkaStreamsCustomizers) { + this.kafkaStreamsCustomizers.addAll(kafkaStreamsCustomizers); + } + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaStreamsCustomizer.java b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaStreamsCustomizer.java new file mode 100644 index 00000000..a3c947fa --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaStreamsCustomizer.java @@ -0,0 +1,35 @@ +/* + * Copyright 2018 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.kafka.core; + +import org.apache.kafka.streams.KafkaStreams; + +/** + * Callback interface that can be used to configure {@link KafkaStreams} directly. + * + * @author Nurettin Yilmaz + * + * @since 2.1.5 + * + * @see StreamsBuilderFactoryBean + */ +@FunctionalInterface +public interface KafkaStreamsCustomizer { + + void customize(KafkaStreams kafkaStreams); + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/StreamsBuilderFactoryBean.java b/spring-kafka/src/main/java/org/springframework/kafka/core/StreamsBuilderFactoryBean.java index c3da9a6a..621431ae 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/StreamsBuilderFactoryBean.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/StreamsBuilderFactoryBean.java @@ -23,6 +23,8 @@ import org.apache.kafka.streams.KafkaClientSupplier; import org.apache.kafka.streams.KafkaStreams; import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.StreamsConfig; +import org.apache.kafka.streams.Topology; +import org.apache.kafka.streams.processor.StateRestoreListener; import org.apache.kafka.streams.processor.internals.DefaultKafkaClientSupplier; import org.springframework.beans.factory.config.AbstractFactoryBean; @@ -35,10 +37,14 @@ import org.springframework.util.Assert; * An {@link AbstractFactoryBean} for the {@link StreamsBuilder} instance * and lifecycle control for the internal {@link KafkaStreams} instance. * + *

A fine grained control on {@link KafkaStreams} can be achieved by + * {@link KafkaStreamsCustomizer}s

+ * * @author Artem Bilan * @author Ivan Ursul * @author Soby Chacko * @author Zach Olauson + * @author Nurettin Yilmaz * * @since 1.1.4 */ @@ -46,24 +52,28 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean props = new HashMap<>(); + props.put(StreamsConfig.APPLICATION_ID_CONFIG, APPLICATION_ID); + props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddresses); + return new StreamsConfig(props); + } + + + private KafkaStreamsCustomizer customizer() { + return kafkaStreams -> kafkaStreams.setStateListener(STATE_LISTENER); + } + + } + + static class TestStateListener implements KafkaStreams.StateListener { + + private KafkaStreams.State currentState; + + @Override + public void onChange(KafkaStreams.State newState, KafkaStreams.State oldState) { + this.currentState = newState; + } + + KafkaStreams.State getCurrentState() { + return this.currentState; + } + + } + +} diff --git a/src/reference/asciidoc/streams.adoc b/src/reference/asciidoc/streams.adoc index b30d3c6c..60f63b6e 100644 --- a/src/reference/asciidoc/streams.adoc +++ b/src/reference/asciidoc/streams.adoc @@ -70,7 +70,9 @@ If you would like to control lifecycle manually (e.g. stop and start by some con Since `StreamsBuilderFactoryBean` utilize its internal `KafkaStreams` instance, it is safe to stop and restart it again - a new `KafkaStreams` is created on each `start()`. Also consider using different `StreamsBuilderFactoryBean` s, if you would like to control lifecycles for `KStream` instances separately. -You can specify `KafkaStreams.StateListener` and `Thread.UncaughtExceptionHandler` options on the `StreamsBuilderFactoryBean` which are delegated to the internal `KafkaStreams` instance. +You also can specify `KafkaStreams.StateListener`, `Thread.UncaughtExceptionHandler` and `StateRestoreListener` options on the `StreamsBuilderFactoryBean` which are delegated to the internal `KafkaStreams` instance. +Also apart from setting those options indirectly on `StreamsBuilderFactoryBean`, starting with _version 2.1.5_, a `KafkaStreamsCustomizer` callback interface can be used to configure inner `KafkaStreams` instance. +Note that `KafkaStreamsCustomizer` will override the options which are given via `StreamsBuilderFactoryBean`. That internal `KafkaStreams` instance can be accessed via `StreamsBuilderFactoryBean.getKafkaStreams()` if you need to perform some `KafkaStreams` operations directly. You can autowire `StreamsBuilderFactoryBean` bean by type, but you should be sure that you use full type in the bean definition, for example: