diff --git a/spring-kafka/src/main/java/org/springframework/kafka/annotation/EnableKafkaStreams.java b/spring-kafka/src/main/java/org/springframework/kafka/annotation/EnableKafkaStreams.java index a9c27cc0..91b1a7ed 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/annotation/EnableKafkaStreams.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/annotation/EnableKafkaStreams.java @@ -23,7 +23,7 @@ import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; import org.springframework.context.annotation.Import; -import org.springframework.kafka.core.KStreamBuilderFactoryBean; +import org.springframework.kafka.core.StreamsBuilderFactoryBean; /** * Enable default Kafka Streams components. To be used on @@ -44,8 +44,8 @@ import org.springframework.kafka.core.KStreamBuilderFactoryBean; * * * That {@link KafkaStreamsDefaultConfiguration#DEFAULT_STREAMS_CONFIG_BEAN_NAME} is required - * to declare {@link KStreamBuilderFactoryBean} with the - * {@link KafkaStreamsDefaultConfiguration#DEFAULT_KSTREAM_BUILDER_BEAN_NAME}. + * to declare {@link StreamsBuilderFactoryBean} with the + * {@link KafkaStreamsDefaultConfiguration#DEFAULT_STREAMS_BUILDER_BEAN_NAME}. *

* Also to enable Kafka Streams feature you should be sure that the {@code kafka-streams} jar is * on classpath. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaStreamsDefaultConfiguration.java b/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaStreamsDefaultConfiguration.java index b7dd7775..73c2acab 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaStreamsDefaultConfiguration.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaStreamsDefaultConfiguration.java @@ -23,10 +23,10 @@ import org.springframework.beans.factory.UnsatisfiedDependencyException; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.kafka.core.KStreamBuilderFactoryBean; +import org.springframework.kafka.core.StreamsBuilderFactoryBean; /** - * {@code @Configuration} class that registers a {@link KStreamBuilderFactoryBean} + * {@code @Configuration} class that registers a {@link StreamsBuilderFactoryBean} * if {@link StreamsConfig} with the name * {@link KafkaStreamsDefaultConfiguration#DEFAULT_STREAMS_CONFIG_BEAN_NAME} is present * in the application context. Otherwise a {@link UnsatisfiedDependencyException} is thrown. @@ -43,25 +43,25 @@ public class KafkaStreamsDefaultConfiguration { /** * The bean name for the {@link StreamsConfig} to be used for the default - * {@link KStreamBuilderFactoryBean} bean definition. + * {@link StreamsBuilderFactoryBean} bean definition. */ public static final String DEFAULT_STREAMS_CONFIG_BEAN_NAME = "defaultKafkaStreamsConfig"; /** - * The bean name for auto-configured default {@link KStreamBuilderFactoryBean}. + * The bean name for auto-configured default {@link StreamsBuilderFactoryBean}. */ - public static final String DEFAULT_KSTREAM_BUILDER_BEAN_NAME = "defaultKStreamBuilder"; + public static final String DEFAULT_STREAMS_BUILDER_BEAN_NAME = "defaultKafkaStreamsBuilder"; - @Bean(name = DEFAULT_KSTREAM_BUILDER_BEAN_NAME) - public KStreamBuilderFactoryBean defaultKStreamBuilder( + @Bean(name = DEFAULT_STREAMS_BUILDER_BEAN_NAME) + public StreamsBuilderFactoryBean defaultKafkaStreamsBuilder( @Qualifier(DEFAULT_STREAMS_CONFIG_BEAN_NAME) ObjectProvider streamsConfigProvider) { StreamsConfig streamsConfig = streamsConfigProvider.getIfAvailable(); if (streamsConfig != null) { - return new KStreamBuilderFactoryBean(streamsConfig); + return new StreamsBuilderFactoryBean(streamsConfig); } else { throw new UnsatisfiedDependencyException(KafkaStreamsDefaultConfiguration.class.getName(), - DEFAULT_KSTREAM_BUILDER_BEAN_NAME, "streamsConfig", "There is no '" + + DEFAULT_STREAMS_BUILDER_BEAN_NAME, "streamsConfig", "There is no '" + DEFAULT_STREAMS_CONFIG_BEAN_NAME + "' StreamsConfig bean in the application context.\n" + "Consider to declare one or don't use @EnableKafkaStreams."); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/KStreamBuilderFactoryBean.java b/spring-kafka/src/main/java/org/springframework/kafka/core/StreamsBuilderFactoryBean.java similarity index 93% rename from spring-kafka/src/main/java/org/springframework/kafka/core/KStreamBuilderFactoryBean.java rename to spring-kafka/src/main/java/org/springframework/kafka/core/StreamsBuilderFactoryBean.java index f5561567..b4578eb0 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/KStreamBuilderFactoryBean.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/StreamsBuilderFactoryBean.java @@ -39,7 +39,7 @@ import org.springframework.util.Assert; * * @since 1.1.4 */ -public class KStreamBuilderFactoryBean extends AbstractFactoryBean implements SmartLifecycle { +public class StreamsBuilderFactoryBean extends AbstractFactoryBean implements SmartLifecycle { private static final int DEFAULT_CLOSE_TIMEOUT = 10; @@ -61,12 +61,12 @@ public class KStreamBuilderFactoryBean extends AbstractFactoryBean streamsConfig) { + public StreamsBuilderFactoryBean(Map streamsConfig) { Assert.notNull(streamsConfig, "'streamsConfig' must not be null"); this.streamsConfig = new StreamsConfig(streamsConfig); } @@ -172,9 +172,9 @@ public class KStreamBuilderFactoryBean extends AbstractFactoryBean resultFuture; @Autowired - private KStreamBuilderFactoryBean kStreamBuilderFactoryBean; + private StreamsBuilderFactoryBean streamsBuilderFactoryBean; @Autowired private KafkaEmbedded kafkaEmbedded; @@ -112,15 +112,15 @@ public class KafkaStreamsTests { assertThat(this.kafkaEmbedded.getKafkaServer(0).config().deleteTopicEnable()).isTrue(); assertThat(this.kafkaEmbedded.getKafkaServer(0).config().brokerId()).isEqualTo(2); - this.kStreamBuilderFactoryBean.stop(); + this.streamsBuilderFactoryBean.stop(); CountDownLatch stateLatch = new CountDownLatch(1); - this.kStreamBuilderFactoryBean.setStateListener((newState, oldState) -> stateLatch.countDown()); + this.streamsBuilderFactoryBean.setStateListener((newState, oldState) -> stateLatch.countDown()); Thread.UncaughtExceptionHandler exceptionHandler = mock(Thread.UncaughtExceptionHandler.class); - this.kStreamBuilderFactoryBean.setUncaughtExceptionHandler(exceptionHandler); + this.streamsBuilderFactoryBean.setUncaughtExceptionHandler(exceptionHandler); - this.kStreamBuilderFactoryBean.start(); + this.streamsBuilderFactoryBean.start(); String payload = "foo" + UUID.randomUUID().toString(); String payload2 = "foo" + UUID.randomUUID().toString(); @@ -137,7 +137,7 @@ public class KafkaStreamsTests { assertThat(stateLatch.await(10, TimeUnit.SECONDS)).isTrue(); - KafkaStreams kafkaStreams = this.kStreamBuilderFactoryBean.getKafkaStreams(); + KafkaStreams kafkaStreams = this.streamsBuilderFactoryBean.getKafkaStreams(); StreamThread[] threads = KafkaTestUtils.getPropertyValue(kafkaStreams, "threads", StreamThread[].class); assertThat(threads).isNotEmpty(); diff --git a/src/reference/asciidoc/streams.adoc b/src/reference/asciidoc/streams.adoc index 38b86f73..09cf80e0 100644 --- a/src/reference/asciidoc/streams.adoc +++ b/src/reference/asciidoc/streams.adoc @@ -17,11 +17,7 @@ The reference Apache Kafka Streams documentation suggests this way of using the // from which input topics to read, which stream operations (filter, map, etc.) // should be called, and so on. -KStreamBuilder builder = ...; // when using the Kafka Streams DSL -// -// OR -// -TopologyBuilder builder = ...; // when using the Processor API +StreamsBuilder builder = ...; // when using the Kafka Streams DSL // Use the configuration to tell your application where the Kafka cluster is, // which serializers/deserializers to use by default, to specify security settings, @@ -37,69 +33,69 @@ streams.start(); streams.close(); ---- -So, we have two main components: `KStreamBuilder` (which extends `TopologyBuilder` as well) with an API to build `KStream` (or `KTable`) instances and `KafkaStreams` to manage their lifecycle. -Note: all `KStream` instances exposed to a `KafkaStreams` instance by a single `KStreamBuilder` will be started and stopped at the same time, even if they have a fully different logic. -In other words all our streams defined by a `KStreamBuilder` are tied with a single lifecycle control. +So, we have two main components: `StreamsBuilder` with an API to build `KStream` (or `KTable`) instances and `KafkaStreams` to manage their lifecycle. +Note: all `KStream` instances exposed to a `KafkaStreams` instance by a single `StreamsBuilder` will be started and stopped at the same time, even if they have a fully different logic. +In other words all our streams defined by a `StreamsBuilder` are tied with a single lifecycle control. Once a `KafkaStreams` instance has been closed via `streams.close()` it cannot be restarted, and a new `KafkaStreams` instance to restart stream processing must be created instead. ==== Spring Management -To simplify the usage of Kafka Streams from the Spring application context perspective and utilize the lifecycle management via container, the Spring for Apache Kafka introduces `KStreamBuilderFactoryBean`. -This is an `AbstractFactoryBean` implementation to expose a `KStreamBuilder` singleton instance as a bean: +To simplify the usage of Kafka Streams from the Spring application context perspective and utilize the lifecycle management via container, the Spring for Apache Kafka introduces `StreamsBuilderFactoryBean`. +This is an `AbstractFactoryBean` implementation to expose a `StreamsBuilder` singleton instance as a bean: [source, java] ---- @Bean -public FactoryBean myKStreamBuilder(StreamsConfig streamsConfig) { - return new KStreamBuilderFactoryBean(streamsConfig); +public FactoryBean myKStreamBuilder(StreamsConfig streamsConfig) { + return new StreamsBuilderFactoryBean(streamsConfig); } ---- -The `KStreamBuilderFactoryBean` also implements `SmartLifecycle` to manage lifecycle of an internal `KafkaStreams` instance. +The `StreamsBuilderFactoryBean` also implements `SmartLifecycle` to manage lifecycle of an internal `KafkaStreams` instance. Similar to the Kafka Streams API, the `KStream` instances must be defined before starting the `KafkaStreams`, and that also applies for the Spring API for Kafka Streams. -Therefore we have to declare `KStream` s on the `KStreamBuilder` before the application context is refreshed, when we use default `autoStartup = true` on the `KStreamBuilderFactoryBean`. +Therefore we have to declare `KStream` s on the `StreamsBuilder` before the application context is refreshed, when we use default `autoStartup = true` on the `StreamsBuilderFactoryBean`. For example, `KStream` can be just as a regular bean definition, meanwhile the Kafka Streams API is used without any impacts: [source, java] ---- @Bean -public KStream kStream(KStreamBuilder kStreamBuilder) { +public KStream kStream(StreamsBuilder kStreamBuilder) { KStream stream = kStreamBuilder.stream(STREAMING_TOPIC1); // Fluent KStream API return stream; } ---- -If you would like to control lifecycle manually (e.g. stop and start by some condition), you can reference the `KStreamBuilderFactoryBean` bean directly using factory bean (`&`) http://docs.spring.io/spring/docs/current/spring-framework-reference/html/beans.html#beans-factory-extension-factorybean[prefix]. -Since `KStreamBuilderFactoryBean` 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 `KStreamBuilderFactoryBean` s, if you would like to control lifecycles for `KStream` instances separately. +If you would like to control lifecycle manually (e.g. stop and start by some condition), you can reference the `StreamsBuilderFactoryBean` bean directly using factory bean (`&`) http://docs.spring.io/spring/docs/current/spring-framework-reference/html/beans.html#beans-factory-extension-factorybean[prefix]. +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 `KStreamBuilderFactoryBean` which are delegated to the internal `KafkaStreams` instance. -That internal `KafkaStreams` instance can be accessed via `KStreamBuilderFactoryBean.getKafkaStreams()` if you need to perform some `KafkaStreams` operations directly. -You can autowire `KStreamBuilderFactoryBean` bean by type, but you should be sure that you use full type in the bean definition, for example: +You can specify `KafkaStreams.StateListener` and `Thread.UncaughtExceptionHandler` options on the `StreamsBuilderFactoryBean` which are delegated to the internal `KafkaStreams` instance. +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: [source,java] ---- @Bean -public KStreamBuilderFactoryBean myKStreamBuilder(StreamsConfig streamsConfig) { - return new KStreamBuilderFactoryBean(streamsConfig); +public StreamsBuilderFactoryBean myKStreamBuilder(StreamsConfig streamsConfig) { + return new StreamsBuilderFactoryBean(streamsConfig); } ... @Autowired -private KStreamBuilderFactoryBean myKStreamBuilderFactoryBean; +private StreamsBuilderFactoryBean myKStreamBuilderFactoryBean; ---- Or add `@Qualifier` for injection by name if you use interface bean definition: [source,java] ---- @Bean -public FactoryBean myKStreamBuilder(StreamsConfig streamsConfig) { - return new KStreamBuilderFactoryBean(streamsConfig); +public FactoryBean myKStreamBuilder(StreamsConfig streamsConfig) { + return new StreamsBuilderFactoryBean(streamsConfig); } ... @Autowired @Qualifier("&myKStreamBuilder") -private KStreamBuilderFactoryBean myKStreamBuilderFactoryBean; +private StreamsBuilderFactoryBean myKStreamBuilderFactoryBean; ---- ==== JSON Serdes @@ -115,13 +111,13 @@ stream.through(Serdes.Integer(), new JsonSerde<>(Foo.class), "foos"); ==== Configuration -To configure the Kafka Streams environment, the `KStreamBuilderFactoryBean` requires a `Map` of particular properties or a `StreamsConfig` instance. +To configure the Kafka Streams environment, the `StreamsBuilderFactoryBean` requires a `Map` of particular properties or a `StreamsConfig` instance. See Apache Kafka https://kafka.apache.org/0102/documentation/#streamsconfigs[documentation] for all possible options. To avoid boilerplate code for most cases, especially when you develop micro services, Spring for Apache Kafka provides the `@EnableKafkaStreams` annotation, which should be placed alongside with `@Configuration`. Only you need is to declare `StreamsConfig` bean with the `defaultKafkaStreamsConfig` name. -A `KStreamBuilder` bean with the `defaultKStreamBuilder` name will be declare in the application context automatically. -Any additional `KStreamBuilderFactoryBean` beans can be declared and used as well. +A `StreamsBuilder` bean with the `defaultKafkaStreamsBuilder` name will be declare in the application context automatically. +Any additional `StreamsBuilderFactoryBean` beans can be declared and used as well. ==== Kafka Streams Example @@ -145,7 +141,7 @@ public static class KafkaStreamsConfiguration { } @Bean - public KStream kStream(KStreamBuilder kStreamBuilder) { + public KStream kStream(StreamsBuilder kStreamBuilder) { KStream stream = kStreamBuilder.stream("streamingTopic1"); stream .mapValues(String::toUpperCase)