Kafka Streams binder docs updates
Refactoring kafka streams docs into a separate module Resolves #293
This commit is contained in:
Soby Chacko
committed by
Oleg Zhurakousky
Oleg Zhurakousky
parent
a5344655cb
commit
cdffcd844c
@@ -0,0 +1,558 @@
|
||||
== Kafka Streams Binding Capabilities of Spring Cloud Stream
|
||||
|
||||
Spring Cloud Stream Kafka support also includes a binder specifically designed for Apache Kafka Streams binding.
|
||||
Using this binder, applications can be written that leverage the Apache Kafka Streams API.
|
||||
For more information on Kafka Streams, see https://kafka.apache.org/documentation/streams/developer-guide[Kafka Streams API Developer Manual]
|
||||
|
||||
Kafka Streams support in Spring Cloud Stream is based on the foundations provided by the Spring Kafka project.
|
||||
For details on that support, see http://docs.spring.io/spring-kafka/reference/html/_reference.html#kafka-streams[Kafaka Streams Support in Spring Kafka].
|
||||
|
||||
Here are the maven coordinates for the Spring Cloud Stream Kafka Streams binder artifact.
|
||||
|
||||
[source,xml]
|
||||
----
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka-streams</artifactId>
|
||||
</dependency>
|
||||
----
|
||||
|
||||
High level streams DSL provided through the Kafka Streams API can be used through Spring Cloud Stream support.
|
||||
Some minimal support for writing applications using the processor API is also available through the binder.
|
||||
Kafka Streams applications using the Spring Cloud Stream support can be written using the processor model, i.e. messages read from an inbound topic and messages written to an outbound topic or using the sink style where it does not have an output binding.
|
||||
|
||||
=== Usage example of high level streams DSL
|
||||
|
||||
This application will listen from a Kafka topic and write the word count for each unique word that it sees in a 5 seconds time window.
|
||||
|
||||
[source]
|
||||
----
|
||||
@SpringBootApplication
|
||||
@EnableBinding(KStreamProcessor.class)
|
||||
public class WordCountProcessorApplication {
|
||||
|
||||
@StreamListener("input")
|
||||
@SendTo("output")
|
||||
public KStream<?, WordCount> process(KStream<?, String> input) {
|
||||
return input
|
||||
.flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
|
||||
.groupBy((key, value) -> value)
|
||||
.windowedBy(TimeWindows.of(5000))
|
||||
.count(Materialized.as("WordCounts-multi"))
|
||||
.toStream()
|
||||
.map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end()))));
|
||||
}
|
||||
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(WordCountProcessorApplication.class, args);
|
||||
}
|
||||
----
|
||||
|
||||
If you build it as a Spring Boot uber jar, you can run the above example in the following way:
|
||||
|
||||
[source]
|
||||
----
|
||||
java -jar uber.jar --spring.cloud.stream.bindings.input.destination=words --spring.cloud.stream.bindings.output.destination=counts
|
||||
----
|
||||
|
||||
This means that the application will listen from the incoming Kafka topic `words` and write to the output topic `counts`.
|
||||
|
||||
Spring Cloud Stream will ensure that the messages from both the incoming and outgoing topics are bound as KStream objects.
|
||||
Applications can exclusively focus on the business aspects of the code, i.e. writing the logic required in the processor rather than setting up the streams specific configuration required by the Kafka Streams infrastructure.
|
||||
All such infrastructure details are handled by the framework.
|
||||
|
||||
=== Multiple Input bindings on the inbound
|
||||
|
||||
Spring Cloud Stream Kafka Streams binder allows the users to write applications with multiple bindings.
|
||||
There are use cases in which you may want to have multiple incoming KStream objects or a combination of KStream and KTable objects.
|
||||
Both of these flavors are supported.
|
||||
Here are some examples.
|
||||
|
||||
[source]
|
||||
----
|
||||
@EnableBinding(KStreamKTableBinding.class)
|
||||
.....
|
||||
.....
|
||||
@StreamListener
|
||||
public void process(@Input("inputStream") KStream<String, PlayEvent> playEvents,
|
||||
@Input("inputTable") KTable<Long, Song> songTable) {
|
||||
....
|
||||
....
|
||||
}
|
||||
|
||||
interface KStreamKTableBinding {
|
||||
|
||||
@Input("inputStream")
|
||||
KStream<?, ?> inputStream();
|
||||
|
||||
@Input("inputTable")
|
||||
KTable<?, ?> inputTable();
|
||||
}
|
||||
|
||||
----
|
||||
|
||||
In the above example, the application is written in a sink style, i.e. there are no output bindings and the application has to make the decision to what needs to happen.
|
||||
Most likely, when you write applications this way, you might want to send the information downstream or store them in a state store (See below for Queryable State Stores).
|
||||
|
||||
In the case of incoming KTable, if you want to materialize it as a state store, you have to express that through the following property.
|
||||
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.kafka.streams.bindings.inputTable.consumer.materializedAs: all-songs
|
||||
----
|
||||
|
||||
Here is an example for multiple input bindings and an output binding (processor style).
|
||||
|
||||
[source]
|
||||
----
|
||||
@EnableBinding(KStreamKTableBinding.class)
|
||||
....
|
||||
....
|
||||
|
||||
@StreamListener
|
||||
@SendTo("output")
|
||||
public KStream<String, Long> process(@Input("input") KStream<String, Long> userClicksStream,
|
||||
@Input("inputTable") KTable<String, String> userRegionsTable) {
|
||||
....
|
||||
....
|
||||
}
|
||||
|
||||
interface KStreamKTableBinding extends KafkaStreamsProcessor {
|
||||
|
||||
@Input("inputX")
|
||||
KTable<?, ?> inputTable();
|
||||
}
|
||||
|
||||
----
|
||||
|
||||
=== Support for branching in Kafka Streams API
|
||||
|
||||
Kafka Streams allow outbound data to be split into multiple topics based on some predicates.
|
||||
Spring Cloud Stream Kafka Streams binder provides support for this feature without losing the overall programming model exposed through `StreamListener` in the end user application.
|
||||
You write the application in the usual way as demonstrated above in the word count example.
|
||||
When using the branching feature, you are required to do a few things.
|
||||
First, you need to make sure that your return type is `KStream[]` instead of a regular `KStream`.
|
||||
Then you need to use the `SendTo` annotation containing the output bindings in the order (example below).
|
||||
For each of these output bindings, you need to configure destination, content-type etc. as required by any other standard Spring Cloud Stream application
|
||||
|
||||
Here is an example:
|
||||
|
||||
[source]
|
||||
----
|
||||
@EnableBinding(KStreamProcessorWithBranches.class)
|
||||
@EnableAutoConfiguration
|
||||
public static class WordCountProcessorApplication {
|
||||
|
||||
@Autowired
|
||||
private TimeWindows timeWindows;
|
||||
|
||||
@StreamListener("input")
|
||||
@SendTo({"output1","output2","output3})
|
||||
public KStream<?, WordCount>[] process(KStream<Object, String> input) {
|
||||
|
||||
Predicate<Object, WordCount> isEnglish = (k, v) -> v.word.equals("english");
|
||||
Predicate<Object, WordCount> isFrench = (k, v) -> v.word.equals("french");
|
||||
Predicate<Object, WordCount> isSpanish = (k, v) -> v.word.equals("spanish");
|
||||
|
||||
return input
|
||||
.flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
|
||||
.groupBy((key, value) -> value)
|
||||
.windowedBy(timeWindows)
|
||||
.count(Materialized.as("WordCounts-1"))
|
||||
.toStream()
|
||||
.map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end()))))
|
||||
.branch(isEnglish, isFrench, isSpanish);
|
||||
}
|
||||
|
||||
interface KStreamProcessorWithBranches {
|
||||
|
||||
@Input("input")
|
||||
KStream<?, ?> input();
|
||||
|
||||
@Output("output1")
|
||||
KStream<?, ?> output1();
|
||||
|
||||
@Output("output2")
|
||||
KStream<?, ?> output2();
|
||||
|
||||
@Output("output3")
|
||||
KStream<?, ?> output3();
|
||||
}
|
||||
}
|
||||
----
|
||||
|
||||
Then in the properties:
|
||||
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.bindings.output1.contentType: application/json
|
||||
spring.cloud.stream.bindings.output2.contentType: application/json
|
||||
spring.cloud.stream.bindings.output3.contentType: application/json
|
||||
spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms: 1000
|
||||
spring.cloud.stream.kafka.streams.binder.configuration:
|
||||
default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
|
||||
default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
|
||||
spring.cloud.stream.bindings.output1:
|
||||
destination: foo
|
||||
producer:
|
||||
headerMode: raw
|
||||
spring.cloud.stream.bindings.output2:
|
||||
destination: bar
|
||||
producer:
|
||||
headerMode: raw
|
||||
spring.cloud.stream.bindings.output3:
|
||||
destination: fox
|
||||
producer:
|
||||
headerMode: raw
|
||||
spring.cloud.stream.bindings.input:
|
||||
destination: words
|
||||
consumer:
|
||||
headerMode: raw
|
||||
----
|
||||
|
||||
=== Message conversion in Spring Cloud Stream Kafka Streams applications
|
||||
|
||||
Spring Cloud Stream Kafka Streams binder allows the usage of usual patterns for content type conversions as in other message channel based binder applications.
|
||||
Many Kafka Streams operations - that are part of the actual application and not at the inbound and outbound - need to know the type of SerDe’s used to correctly transform key and value data.
|
||||
Therefore, it may be more natural to rely on the SerDe facilities provided by the Apache Kafka Streams library itself for inbound and outbound conversions rather than using the content type conversions offered by the framework.
|
||||
On the other hand, you might be already familiar with the content type conversion patterns in spring cloud stream and want to keep using them for inbound and outbound conversions.
|
||||
Both options are supported in the Spring Cloud Stream binder for Apache Kafka Streams.
|
||||
|
||||
==== Outbound serialization
|
||||
|
||||
If native encoding is disabled (which is the default), then the framework will convert the message using the contentType set by the user (or the default content type of application/json).
|
||||
It will ignore any Serde set on the outbound in this case for outbound serialization.
|
||||
|
||||
Here is the property to set the contentType on the outbound.
|
||||
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.bindings.output.contentType: application/json
|
||||
----
|
||||
|
||||
Here is the property to enable native encoding.
|
||||
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.bindings.output.nativeEncoding: true
|
||||
----
|
||||
|
||||
If native encoding is enabled on the output binding (user has to explicitly enable it as above), then the framework will skip doing any message conversion on the outbound.
|
||||
In that case, it will use the Serde set by the user.
|
||||
First, it checks for the `valueSerde` property set on the actual output binding. Here is an example
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde: org.apache.kafka.common.serialization.Serdes$StringSerde
|
||||
----
|
||||
If this property is not set, then it will default to the common value Serde - `spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde`.
|
||||
|
||||
It is worth to mention that Spring Cloud Stream Kafka Streams binder does not serialize the keys on outbound, rather it is always done by Kafka itself.
|
||||
Therefore, you either have to specify the keySerde property on the binding or it will default to the application wide common keySerde set on the streams configuration.
|
||||
|
||||
Binding level key serde:
|
||||
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde
|
||||
----
|
||||
|
||||
Common Key serde:
|
||||
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde
|
||||
----
|
||||
|
||||
If branching is used, then you need to use multiple output bindings. For example,
|
||||
|
||||
[source]
|
||||
----
|
||||
interface KStreamProcessorWithBranches {
|
||||
|
||||
@Input("input")
|
||||
KStream<?, ?> input();
|
||||
|
||||
@Output("output1")
|
||||
KStream<?, ?> output1();
|
||||
|
||||
@Output("output2")
|
||||
KStream<?, ?> output2();
|
||||
|
||||
@Output("output3")
|
||||
KStream<?, ?> output3();
|
||||
}
|
||||
----
|
||||
|
||||
If nativeEncoding is set, then you can set different Serde values on these individual output bindings as below.
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.kstream.bindings.output1.producer.valueSerde=IntegerSerde
|
||||
spring.cloud.stream.kstream.bindings.outpu2t.producer.valueSerde=StringSerde
|
||||
spring.cloud.stream.kstream.bindings.output3.producer.valueSerde=JsonSerde
|
||||
----
|
||||
|
||||
Then if you have `SendTo` like this, @SendTo({"output1", "output2", "output3"}), the `KStream[]` from the branches are applied with proper Serde objects as defined above.
|
||||
If you are not enabling nativeEncoding, you can then set different contentType values on the output bindings as below.
|
||||
In that case, the framework will use the appropriate message converter to convert the messages before sending to Kafka.
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.bindings.output1.contentType: application/json
|
||||
spring.cloud.stream.bindings.output2.contentType: application/java-serialzied-object
|
||||
spring.cloud.stream.bindings.output3.contentType: application/octet-stream
|
||||
----
|
||||
|
||||
==== Inbound Deserialization
|
||||
|
||||
Similar rules apply to data deserialization on the inbound as in the case of outbound serialization.
|
||||
|
||||
If native decoding is disabled (which is the default), then the framework will convert the message using the contentType set by the user (or the default content type of application/json).
|
||||
It will ignore any Serde set on the inbound in this case for inbound dserialization.
|
||||
|
||||
Here is the property to set the contentType on the inbound.
|
||||
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.bindings.input.contentType: application/json
|
||||
----
|
||||
|
||||
Here is the property to enable native decoding.
|
||||
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.bindings.input.nativeDecoding: true
|
||||
----
|
||||
|
||||
If native decoding is enabled on the input binding (user has to explicitly enable it as above), then the framework will skip doing any message conversion on the inbound.
|
||||
In that case, it will use the Serde set by the user.
|
||||
First, it checks for the `valueSerde` property set on the actual input binding. Here is an example
|
||||
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde: org.apache.kafka.common.serialization.Serdes$StringSerde
|
||||
----
|
||||
If this property is not set, then it will default to the common value Serde - `spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde`.
|
||||
|
||||
It is worth to mention that Spring Cloud Stream Kafka Streams binder does not deserialize the keys on inbound, rather it is always done by Kafka itself.
|
||||
Therefore, you either have to specify the keySerde property on the binding or it will default to the application wide common keySerde set on the streams configuration.
|
||||
|
||||
Binding level key serde:
|
||||
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.kafka.streams.bindings.input.consumer.keySerde
|
||||
----
|
||||
|
||||
Common Key serde:
|
||||
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde
|
||||
----
|
||||
|
||||
As in the case of KStream branching on the outbound, the benefit of setting value Serde per binding is that if you have multiple input bindings (multiple KStreams) and they all require separate value Serdes, then you can configure them individually.
|
||||
If you use the common configuration approach, then that is not possible.
|
||||
|
||||
==== Error handling on Deserialization exceptions
|
||||
|
||||
Apache Kafka Streams now provide the capability for natively handling exceptions from deserialization errors.
|
||||
For details on this support, please see https://cwiki.apache.org/confluence/display/KAFKA/KIP-161%3A+streams+deserialization+exception+handlers[this]
|
||||
Out of the box, Apache Kafka Streams provide two kinds of deserialization exception handlers - logAndContinue and logAndFail.
|
||||
As the name indicates, the former will log the error and continue processing next records and the latter will log the error and fai..
|
||||
LogAndFail is the default deserialization exception handler.
|
||||
|
||||
Spring Cloud Stream binder for Apache Kafka Streams allows to specify these exception handlers through the following properties.
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.kafka.streams.binder.serdeError: logAndContinue
|
||||
----
|
||||
|
||||
In addition to the above two deserialization exception handlers, the binder also provides a third one for sending the bad records (poison pills) to a DLQ topic.
|
||||
Here is how you enable this DLQ exception handler.
|
||||
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.kafka.streams.binder.serdeError: sendToDlq
|
||||
----
|
||||
When the above property is set, then all records in error from deserialization are sent to the DLQ topic.
|
||||
First it checks, if there is a `dlqName` property is set on the binding itself using the following property.
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.kafka.streams.bindings.input.consumer.dlqName: foo-dlq
|
||||
----
|
||||
If this is set, then the records in error are sent to the topic `foo-dlq`.
|
||||
If this is not set, then it will create a DLQ topic called `error.<input-topic-name>.<group-name>`.
|
||||
|
||||
A couple of things to keep in mind when using the exception handling feature through Spring Cloud Stream binder for Apache Kafka Streams.
|
||||
|
||||
* The property `spring.cloud.stream.kafka.streams.binder.serdeError` is applicable for the entire application.
|
||||
This implies that if there are multiple `StreamListener` methods in the same application, this property is applied to all of them.
|
||||
* The exception handling for deserialization works consistently with native deserialization and framework provided message conversion.
|
||||
|
||||
==== Handling Non-Deserialization exceptions
|
||||
|
||||
Other kinds of error handling is limited in Apache Kafka Streams currently and it is up to the end user applications to handle any such application level errors.
|
||||
One side effect of providing a DLQ for deserialization exception handlers as above is that, it provides a way to get access to the DLQ sending bean directly from your application.
|
||||
Once you get access to that bean, you can programmatically send any exception records from your application to the DLQ.
|
||||
Here is an example for how you may do that.
|
||||
Keep in mind that, this approach only works out of the box when you use the low level processor API in your application as below.
|
||||
It still remains hard to achieve the same using the high level DSL without the library natively providing error handling support, but this example provides some hints to work around.
|
||||
|
||||
[source]
|
||||
----
|
||||
@Autowired
|
||||
private SendToDlqAndContinue dlqHandler;
|
||||
|
||||
@StreamListener("input")
|
||||
@SendTo("output")
|
||||
public KStream<?, WordCount> process(KStream<Object, String> input) {
|
||||
|
||||
input.process(() -> new Processor() {
|
||||
ProcessorContext context;
|
||||
|
||||
@Override
|
||||
public void init(ProcessorContext context) {
|
||||
this.context = context;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void process(Object o, Object o2) {
|
||||
|
||||
try {
|
||||
.....
|
||||
.....
|
||||
}
|
||||
catch(Exception e) {
|
||||
//explicitly provide the kafka topic corresponding to the input binding as the first argument.
|
||||
//DLQ handler will correctly map to the dlq topic from the actual incoming destination.
|
||||
dlqHandler.sendToDlq("topic-name", (byte[]) o1, (byte[]) o2, context.partition());
|
||||
}
|
||||
}
|
||||
|
||||
.....
|
||||
.....
|
||||
});
|
||||
}
|
||||
----
|
||||
|
||||
=== Support for interactive queries
|
||||
|
||||
As part of the public API of the binder, it now exposes a class called `QueryableStoreRegistry`.
|
||||
You can access this as a Spring bean in your application.
|
||||
One easy way to get access to this bean from your application is to autowire the bean as below.
|
||||
|
||||
[source]
|
||||
----
|
||||
@Autowired
|
||||
private QueryableStoreRegistry queryableStoreRegistry;
|
||||
----
|
||||
|
||||
Once you gain access to this bean, then you can find out the particular state store that you are interested in.
|
||||
Here is an example:
|
||||
|
||||
[source]
|
||||
----
|
||||
ReadOnlyKeyValueStore<Object, Object> keyValueStore =
|
||||
queryableStoreRegistry.getQueryableStoreType("my-store", QueryableStoreTypes.keyValueStore());
|
||||
----
|
||||
Then you can retrieve the data that you stored in this store during the execution of your application.
|
||||
|
||||
=== Kafka Streams properties
|
||||
|
||||
We covered all the relevant properties that you need when writing Kafka Streams applications using Spring Cloud Stream, scattered in the above sections, but here they are again.
|
||||
|
||||
The following properties are available at the binder level and must be prefixed with `spring.cloud.stream.kafka.binder.`.
|
||||
|
||||
configuration::
|
||||
Map with a key/value pair containing properties pertaining to Apache Kafka Streams API.
|
||||
This property must be prefixed with `spring.cloud.stream.kafka.streams.binder.`.
|
||||
Following are some examples of using this property.
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
|
||||
spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
|
||||
spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000
|
||||
----
|
||||
|
||||
For more information about all the properties that may go into streams configuration, see StreamsConfig JavaDocs in Apache Kafka Streams docs.
|
||||
|
||||
brokers::
|
||||
Broker URL
|
||||
+
|
||||
Default: `localhost`
|
||||
zkNodes::
|
||||
Zookeeper URL
|
||||
+
|
||||
Default: `localhost`
|
||||
serdeError::
|
||||
Deserialization error handler type.
|
||||
Possible values are - `logAndContinue`, `logAndFail` or `sendToDlq`
|
||||
+
|
||||
Default: `logAndFail`
|
||||
applicationId::
|
||||
Application ID for all the stream configurations in the current application context.
|
||||
You can override the application id for an individual `StreamListener` method using the `group` property on the binding.
|
||||
You have to ensure that you are using the same group name for all input bindings in the case of multiple inputs on the same methods.
|
||||
+
|
||||
Default: `default`
|
||||
|
||||
|
||||
The following properties are available for Kafka Streams producers only and must be prefixed with `spring.cloud.stream.kafka.streams.bindings.<binding name>.producer.`.
|
||||
|
||||
keySerde::
|
||||
key serde to use
|
||||
+
|
||||
Default: `none`.
|
||||
valueSerde::
|
||||
value serde to use
|
||||
+
|
||||
Default: `none`.
|
||||
useNativeEncoding::
|
||||
flag to enable native encoding
|
||||
+
|
||||
Default: `false`.
|
||||
|
||||
The following properties are available for Kafka Streams consumers only and must be prefixed with `spring.cloud.stream.kafka.streams.bindings.<binding name>.consumer.`.
|
||||
|
||||
keySerde::
|
||||
key serde to use
|
||||
+
|
||||
Default: `none`.
|
||||
valueSerde::
|
||||
value serde to use
|
||||
+
|
||||
Default: `none`.
|
||||
materializedAs::
|
||||
state store to materialize when using incoming KTable types
|
||||
+
|
||||
Default: `none`.
|
||||
useNativeDecoding::
|
||||
flag to enable native decoding
|
||||
+
|
||||
Default: `false`.
|
||||
dlqName::
|
||||
DLQ topic name.
|
||||
+
|
||||
Default: `none`.
|
||||
|
||||
Other common properties used from core Spring Cloud Stream.
|
||||
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.bindings.<binding name>.destination
|
||||
spring.cloud.stream.bindings.<binding name>.group
|
||||
----
|
||||
|
||||
TimeWindow properties:
|
||||
|
||||
Windowing is an important concept in stream processing applications.
|
||||
Following properties are available for configuring time windows.
|
||||
|
||||
spring.cloud.stream.kafka.streams.timeWindow.length::
|
||||
When this property is given, you can autowire a `TimeWindows` bean into the application.
|
||||
The value is expressed in milliseconds.
|
||||
+
|
||||
Default: `none`.
|
||||
spring.cloud.stream.kstream.timeWindow.advanceBy::
|
||||
Value is given in milliseconds.
|
||||
+
|
||||
Default: `none`.
|
||||
@@ -518,220 +518,6 @@ On the other hand, if auto topic creation is disabled on the server, then care m
|
||||
|
||||
If you want to have full control over how partitions are allocated, then leave the default settings as they are, i.e. do not exclude the kafka broker jar and ensure that `spring.cloud.stream.kafka.binder.autoCreateTopics` is set to `true`, which is the default.
|
||||
|
||||
== Kafka Streams Binding Capabilities of Spring Cloud Stream
|
||||
|
||||
Spring Cloud Stream Kafka support also includes a binder specifically designed for Kafka Streams binding.
|
||||
Using this binder, applications can be written that leverage the Kafka Streams API.
|
||||
For more information on Kafka Streams, see https://kafka.apache.org/documentation/streams/developer-guide[Kafka Streams API Developer Manual]
|
||||
|
||||
Kafka Streams support in Spring Cloud Stream is based on the foundations provided by the Spring Kafka project.
|
||||
For details on that support, see http://docs.spring.io/spring-kafka/reference/html/_reference.html#kafka-streams[Kafaka Streams Support in Spring Kafka].
|
||||
|
||||
Here are the maven coordinates for the Spring Cloud Stream KStream binder artifact.
|
||||
|
||||
[source,xml]
|
||||
----
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kstream</artifactId>
|
||||
</dependency>
|
||||
----
|
||||
|
||||
High level streams DSL provided through the Kafka Streams API can be used through Spring Cloud Stream support.
|
||||
Kafka Streams applications using the Spring Cloud Stream support can only be written using the processor model, i.e. messages read from an inbound topic and messages written to an outbound topic.
|
||||
|
||||
=== Usage example of high level streams DSL
|
||||
|
||||
This application will listen from a Kafka topic and write the word count for each unique word that it sees in a 5 seconds time window.
|
||||
|
||||
[source]
|
||||
----
|
||||
@SpringBootApplication
|
||||
@EnableBinding(KStreamProcessor.class)
|
||||
public class WordCountProcessorApplication {
|
||||
|
||||
@StreamListener("input")
|
||||
@SendTo("output")
|
||||
public KStream<?, String> process(KStream<?, String> input) {
|
||||
return input
|
||||
.flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
|
||||
.groupBy((key, value) -> value)
|
||||
.windowedBy(TimeWindows.of(5000))
|
||||
.count(Materialized.as("WordCounts-multi"))
|
||||
.toStream()
|
||||
.map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end()))));
|
||||
}
|
||||
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(WordCountProcessorApplication.class, args);
|
||||
}
|
||||
----
|
||||
|
||||
If you build it as a Spring Boot uber jar, you can run the above example in the following way:
|
||||
|
||||
[source]
|
||||
----
|
||||
java -jar uber.jar --spring.cloud.stream.bindings.input.destination=words --spring.cloud.stream.bindings.output.destination=counts
|
||||
----
|
||||
|
||||
This means that the application will listen from the incoming Kafka topic `words` and write to the output topic `counts`.
|
||||
|
||||
Spring Cloud Stream will ensure that the messages from both the incoming and outgoing topics are bound as KStream objects.
|
||||
Applications can exclusively focus on the business aspects of the code, i.e. writing the logic required in the processor rather than setting up the streams specific configuration required by the Kafka Streams infrastructure.
|
||||
All such interactions are handled by the framework.
|
||||
|
||||
=== Message conversion in Spring Cloud Stream Kafka Streams applications
|
||||
|
||||
If the following property is set (default is false), the framework skips all message conversions on the outbound (producer) side and it is then done by Kafka itself.
|
||||
|
||||
`spring.cloud.stream.bindings.output.producer.useNativeEncoding`.
|
||||
|
||||
Similarly, if the following property is set (default is false), any message conversion is skipped on the inbound (consumer) side and natively done by Kafka.
|
||||
|
||||
`spring.cloud.stream.bindings.input.consumer.useNativeDecoding`.
|
||||
|
||||
When native encoding is disabled, then the messages on the outbound are converted by Spring Cloud Stream using the provided `contentType` header.
|
||||
If no contentType is set by the application, it defaults to `application/json`.
|
||||
|
||||
By default, all the out of the box message converters, serialize the data as `byte[]` encoding the proper contentType.
|
||||
In most situations, this is what you want to do, but if other formats than `byte[]` are desired, then ann appropriate message converter needs to be registered in the context and corresponding contentType specified as a property.
|
||||
When doing this way, Serdes should be overridden on the producer using the following property.
|
||||
|
||||
`spring.cloud.stream.kstream.bindings.output.producer.valueSerde`.
|
||||
|
||||
Keys will not get converted, but if the Serdes are different for keys from what is given as the common Serde, you can override that using the following property.
|
||||
|
||||
`spring.cloud.stream.kstream.bindings.output.producer.keySerde`.
|
||||
|
||||
=== Support for branching in Kafka Streams API
|
||||
|
||||
Kafka Streams allow outbound data to be split into multiple topics based on some predicates.
|
||||
Spring Cloud Stream Kafka Streams binder provides support for this feature without losing the overall programming model exposed through `StreamListener` in the end user application.
|
||||
You write the application in the usual way as demonstrated above in the word count example.
|
||||
When using the branching feature, you are required to do a few things.
|
||||
First, you need to make sure that your return type is `KStream[]` instead of a regular `KStream`.
|
||||
Then you need to use the `SendTo` annotation containing the output bindings in the order (example below).
|
||||
For each of these output bindings, you need to configure destination, content-type etc. as required by any other standard Spring Cloud Stream application
|
||||
|
||||
Here is an example:
|
||||
|
||||
[source]
|
||||
----
|
||||
@EnableBinding(KStreamProcessorWithBranches.class)
|
||||
@EnableAutoConfiguration
|
||||
public static class WordCountProcessorApplication {
|
||||
|
||||
@Autowired
|
||||
private TimeWindows timeWindows;
|
||||
|
||||
@StreamListener("input")
|
||||
@SendTo({"output1","output2","output3})
|
||||
public KStream<?, WordCount>[] process(KStream<Object, String> input) {
|
||||
|
||||
Predicate<Object, WordCount> isEnglish = (k, v) -> v.word.equals("english");
|
||||
Predicate<Object, WordCount> isFrench = (k, v) -> v.word.equals("french");
|
||||
Predicate<Object, WordCount> isSpanish = (k, v) -> v.word.equals("spanish");
|
||||
|
||||
return input
|
||||
.flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
|
||||
.groupBy((key, value) -> value)
|
||||
.windowedBy(timeWindows)
|
||||
.count(Materialized.as("WordCounts-1"))
|
||||
.toStream()
|
||||
.map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end()))))
|
||||
.branch(isEnglish, isFrench, isSpanish);
|
||||
}
|
||||
|
||||
interface KStreamProcessorWithBranches {
|
||||
|
||||
@Input("input")
|
||||
KStream<?, ?> input();
|
||||
|
||||
@Output("output1")
|
||||
KStream<?, ?> output1();
|
||||
|
||||
@Output("output2")
|
||||
KStream<?, ?> output2();
|
||||
|
||||
@Output("output3")
|
||||
KStream<?, ?> output3();
|
||||
}
|
||||
}
|
||||
----
|
||||
|
||||
Then in the properties:
|
||||
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.bindings.output1.contentType: application/json
|
||||
spring.cloud.stream.bindings.output2.contentType: application/json
|
||||
spring.cloud.stream.bindings.output3.contentType: application/json
|
||||
spring.cloud.stream.kstream.binder.configuration.commit.interval.ms: 1000
|
||||
spring.cloud.stream.kstream.binder.configuration:
|
||||
key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
|
||||
value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
|
||||
spring.cloud.stream.bindings.output1:
|
||||
destination: foo
|
||||
producer:
|
||||
headerMode: raw
|
||||
spring.cloud.stream.bindings.output2:
|
||||
destination: bar
|
||||
producer:
|
||||
headerMode: raw
|
||||
spring.cloud.stream.bindings.output3:
|
||||
destination: fox
|
||||
producer:
|
||||
headerMode: raw
|
||||
spring.cloud.stream.bindings.input:
|
||||
destination: words
|
||||
consumer:
|
||||
headerMode: raw
|
||||
----
|
||||
|
||||
=== Support for interactive queries
|
||||
|
||||
If access to the `KafkaStreams` is needed for interactive queries, the internal `KafkaStreams` instance can be accessed via `KStreamBuilderFactoryBean.getKafkaStreams()`.
|
||||
You can autowire the `KStreamBuilderFactoryBean` instance provided by the KStream binder.
|
||||
Then you get `KafkaStreams` instance from it and retrieve the underlying store, execute queries on it, etc.
|
||||
|
||||
=== Kafka Streams properties
|
||||
|
||||
configuration::
|
||||
Map with a key/value pair containing properties pertaining to Kafka Streams API.
|
||||
This property must be prefixed with `spring.cloud.stream.kstream.binder.`.
|
||||
|
||||
Following are some examples of using this property.
|
||||
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.kstream.binder.configuration.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
|
||||
spring.cloud.stream.kstream.binder.configuration.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
|
||||
spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000
|
||||
----
|
||||
|
||||
For more information about all the properties that may go into streams configuration, see StreamsConfig JavaDocs.
|
||||
|
||||
There can also be binding specific properties.
|
||||
|
||||
For instance, you can use a different Serde for your input or output destination.
|
||||
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.kstream.bindings.output.producer.keySerde=org.apache.kafka.common.serialization.Serdes$IntegerSerde
|
||||
spring.cloud.stream.kstream.bindings.output.producer.valueSerde=org.apache.kafka.common.serialization.Serdes$LongSerde
|
||||
----
|
||||
|
||||
TimeWindow properties:
|
||||
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.kstream.timeWindow.length (milliseconds)
|
||||
|
||||
When this property is given, you can autowire a `TimeWindows` bean into the application.
|
||||
|
||||
spring.cloud.stream.kstream.timeWindow.advanceBy (milliseconds)
|
||||
----
|
||||
|
||||
[[kafka-error-channels]]
|
||||
== Error Channels
|
||||
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
include::overview.adoc[leveloffset=+1]
|
||||
include::dlq.adoc[leveloffset=+1]
|
||||
include::partitions.adoc[leveloffset=+1]
|
||||
include::kafka-streams.adoc[leveloffset=+1]
|
||||
|
||||
Reference in New Issue
Block a user