Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-479: Update Kafka Streams API and Docs

Browse Source 
Fixes: spring-projects/spring-kafka#479
This commit is contained in:
Artem Bilan
2017-11-29 16:54:53 -05:00
committed by Gary Russell
parent d2ce4aba1e
commit 4c1a8ef5d5
5 changed files with 51 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-kafka/src/main/java/org/springframework/kafka/annotation/EnableKafkaStreams.java
View File

@@ -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;
* </pre>
*
* 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}.
* <p>
* Also to enable Kafka Streams feature you should be sure that the {@code kafka-streams} jar is
* on classpath.

18
spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaStreamsDefaultConfiguration.java
View File

@@ -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<StreamsConfig> 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.");
}

10
spring-kafka/src/main/java/org/springframework/kafka/core/KStreamBuilderFactoryBean.java → spring-kafka/src/main/java/org/springframework/kafka/core/StreamsBuilderFactoryBean.java
View File

@@ -39,7 +39,7 @@ import org.springframework.util.Assert;
*
* @since 1.1.4
*/
public class KStreamBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilder> implements SmartLifecycle {
public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilder> implements SmartLifecycle {
private static final int DEFAULT_CLOSE_TIMEOUT = 10;
@@ -61,12 +61,12 @@ public class KStreamBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
private volatile boolean running;
public KStreamBuilderFactoryBean(StreamsConfig streamsConfig) {
public StreamsBuilderFactoryBean(StreamsConfig streamsConfig) {
Assert.notNull(streamsConfig, "'streamsConfig' must not be null");
this.streamsConfig = streamsConfig;
}
public KStreamBuilderFactoryBean(Map<String, Object> streamsConfig) {
public StreamsBuilderFactoryBean(Map<String, Object> streamsConfig) {
Assert.notNull(streamsConfig, "'streamsConfig' must not be null");
this.streamsConfig = new StreamsConfig(streamsConfig);
}
@@ -172,9 +172,9 @@ public class KStreamBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
}
/**
* Get a managed by this {@link KStreamBuilderFactoryBean} {@link KafkaStreams} instance.
* Get a managed by this {@link StreamsBuilderFactoryBean} {@link KafkaStreams} instance.
* @return KafkaStreams managed instance;
* may be null if this {@link KStreamBuilderFactoryBean} hasn't been started.
* may be null if this {@link StreamsBuilderFactoryBean} hasn't been started.
* @since 1.1.4
*/
public KafkaStreams getKafkaStreams() {

14
spring-kafka/src/test/java/org/springframework/kafka/kstream/KafkaStreamsTests.java
View File

@@ -54,9 +54,9 @@ import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
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.core.ProducerFactory;
import org.springframework.kafka.core.StreamsBuilderFactoryBean;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.support.serializer.JsonSerde;
import org.springframework.kafka.test.context.EmbeddedKafka;
@@ -100,7 +100,7 @@ public class KafkaStreamsTests {
private SettableListenableFuture<String> 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();

58
src/reference/asciidoc/streams.adoc
View File

@@ -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<KStreamBuilder> myKStreamBuilder(StreamsConfig streamsConfig) {
return new KStreamBuilderFactoryBean(streamsConfig);
public FactoryBean<StreamsBuilderFactoryBean> 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<Integer, String> 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<KStreamBuilder> myKStreamBuilder(StreamsConfig streamsConfig) {
return new KStreamBuilderFactoryBean(streamsConfig);
public FactoryBean<StreamsBuilder> 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<Integer, String> kStream(KStreamBuilder kStreamBuilder) {
public KStream<Integer, String> kStream(StreamsBuilder kStreamBuilder) {
KStream<Integer, String> stream = kStreamBuilder.stream("streamingTopic1");
stream
.mapValues(String::toUpperCase)