The following properties are available for Kafka consumers only and must be prefixed with spring.cloud.stream.kafka.bindings.<channelName>.consumer..

The following properties are available for Kafka producers only and must be prefixed with spring.cloud.stream.kafka.bindings.<channelName>.producer..

ResultMetadata meta = sendResultMsg.getHeaders().get(KafkaHeaders.RECORD_METADATA, RecordMetadata.class)

Failed sends go the producer error channel (if configured); see Error Channels. Default: null

+

In this section, we show the use of the preceding properties for specific scenarios.

Starting with Spring Cloud Stream 3.0, Kafka Streams binder allows the applications to be designed and developed using the functional programming style that is available in Java 8. This means that the applications can be concisely represented as a lambda expression of types java.util.function.Function or java.util.function.Consumer.

Let’s take a very basic example.

Albeit simple, this is a complete standalone Spring Boot application that is leveraging Kafka Streams for stream processing. This is a consumer application with no outbound binding and only a single inbound binding. The application consumes data and it simply logs the transformation as standard output. The application contains the SpringBootApplication annotation and a method that is marked as Bean. The bean method is of type java.util.function.Consumer which is parameterized with KStream. Then in the implementation, we are returning a Consumer object that is essentially a lambda expression. Inside the lambda expression, the code for processing the data is provided.

In this application, there is a single input binding that is of type KStream. The binder creates this binding for the application with a name process_in, i.e. the name of the function bean name followed by an underscore and the literal in. You use this binding name to set other properties such as destination. For example, spring.cloud.stream.bindings.process_in.destinaion=my-topic.

Once built as a uber-jar (e.g., kstream-consumer-app.jar), you can run the above example like the following.

Here is another example, where it is a full processor with both input and output bindings. This is the classic word-count example in which the application receives data from a topic, the number of occurrences for each word is then computed in a tumbling time-window.

Here again, this is a complete Spring Boot application. The difference here from the first application, though, the bean method is of type java.util.function.Function. The first parameterized type for the Function is for the input KStream and the second one is for the output. In the method body, a lambda expression is provided that is of type Function and as implementation, the actual business logic is given. Similar to the previously discussed Consumer based application, the input binding here is named as process_in by default. For the output, the binding name is automatically also set to process_out.

Once built as a uber-jar (e.g., wordcount-processor.jar), you can run the above example like the following.

This application will consume messages from the Kafka topic words and the computed results are published to an output topic counts.

Spring Cloud Stream will ensure that the messages from both the incoming and outgoing topics are automatically bound as KStream objects. As a developer, you can exclusively focus on the business aspects of the code, i.e. writing the logic required in the processor. Setting up the Streams DSL specific configuration required by the Kafka Streams infrastructure is automatically handled by the framework.

The two examples we saw above have a single KStream input binding. In both cases, the bindings received the records from a single topic. If you want to multiplex multiple topics into a single KStream binding, you can provide comma separated Kafka topics as destinations below.

spring.cloud.stream.bindings.process_in.destination=topic-1,topic-2,topic-3

Although the functional programming model outlined above is the preferred approach, you can still use the classic StreamListener based approach if you prefer.

Here are some examples.

Following is the equivalent of the Word count example using StreamListener.

As you can see, this is a bit more verbose since you need to provide EnableBinding and the other extra annotations like StreamListener and SendTo to make it a complete application. EnableBinding is where you specify your binding interface that contains your bindings. In this case, we are using the stock KafkaStreamsProcessor binding interface that has the following contracts.

Binder will create bindings for the input KStream and output KStream since you are using a binding interface that contains those declarations.

In addition to the obvious differences in the programming model offered in the functional style, one particular thing that needs to be mentioned here is that the binding names are what you specify in the binding interface. For example, in the above application, since we are using KafkaStreamsProcessor, the binding names are input and output. Binding properties need to use those names. For instance spring.cloud.stream.bindings.input.destination, spring.cloud.stream.bindings.output.destination etc. Keep in mind that this is fundamentally different from the functional style since there the binder generates binding names for the application. This is because the application does not provide any binding interfaces in the functional model using EnableBinding.

Here is another example of a sink where we have two inputs.

Following is the StreamListener equivalent of the same BiFunction based processor that we saw above.

Finally, here is the StreamListener equivalent of the application with three inputs and curried functions.

You might notice that the above two examples are even more verbose since in addition to provide EnableBinding, you also need to write your own custom binding interface as well. Using the functional model, you can avoid all those ceremonial details.

Before we move on looking at the general programming model offered by Kafka Streams binder, here is the StreamListener version of multiple output bindings.

To recap, we have reviewed the various programming model choices when using the Kafka Streams binder.

The binder provides binding capabilities for KStream, KTable and GlobalKTable on the input. KTable and GlobalKTable bindings are only available on the input. Binder supports both input and output bindings for KStream.

The upshot of the programming model of Kafka Streams binder is that the binder provides you the flexibility of going with a fully functional programming model or using the StreamListener based imperative approach.

Application id is a mandatory property that you need to provide for a Kafka Streams application. Spring Cloud Stream Kafka Streams binder allows you to configure this application id in multiple ways.

If you only have one single processor in the application, then you can set this at the binder level using the following property:

spring.cloud.stream.kafka.streams.binder.applicationId.

As a convenience, if you only have a single processor, you can also use spring.application.name as the property to delegate the application id.

If you have multiple Kafka Streams processors in the application, then you need to set the application id per processor. You can set this on the input binding on each processor.

For e.g. imagine that you have to two following functions.

and

Then you must set the application id for each, using the following binding properties.

spring.cloud.stream.kafka.streams.bindings.process_in.applicationId

and

spring.cloud.stream.kafka.streams.bindings.anotherProcess_in.applicationId

For production deployments, it is highly recommended to explicitly specify the application ID through configuration. This is especially going to be very critical if you are auto scaling your application in which case you need to make sure that you are deploying each instance with the same application ID.

If the application does not provide an application ID, then in that case the binder will auto generate a random application ID for you. This is convenient in development scenarios as it avoids the need for explicitly providing the application ID. Please keep in mind that when you rely on this, each time you start the application, it starts with a brand new application id. In the case of functional model, the generated application ID will be the function bean name followed by a UUID which is then postfixed with the literal applicationID. In the case of StreamListener, instead of using the function bean name, the generated application ID will be use the containing class name followed by the method name.

By default, the binder uses the strategy discussed out above to generate the binding name when using the functional style, i.e. <function-bean-name>_<in>|<out>_[0..n], for e.g. process_in, process_in_0 etc. If you want to override those binding names, you can do that by specifying the following properties.

spring.cloud.stream.function.inputBindings.<function-bean-name>.

For e.g. lets say, you have this function.

Binder will generate bindings with names, process_in_0, process_in_1 and process_out. Now, if you want to change them to something else completely, maybe more domain specific binding names. You can do so, as below.

springc.cloud.stream.function.inputBindings.process=users,regions

and

spring.cloud.stream.function.outputBindings.process=clicks

After that, you must set all the binding level properties on these new binding names.

Please keep in mind that with the functional programming model described above, sticking with the default binding names make sense in most situations. The only reason you may still want to do this overriding is when you have larger number of configuration properties and you want to map the bindings to something more domain friendly.

When running Kafka Streams applications, you must provide the Kafka broker server information. If you don’t provide this information, the binder expects that you are running the broker at the default localhost:9092. If that is not the case, then you need to override that. There are a couple of ways to do that.

When it comes to the binder level property, it doesn’t matter if you use the broker property provided through the regular Kafka binder - spring.cloud.stream.kafka.binder.brokers. Kafka Streams binder will first check if Kafka Streams binder specific broker property is set (spring.cloud.stream.kafka.streams.binder.brokers) and if nothing found, it looks for spring.cloud.stream.kafka.binder.brokers.

Keys are always deserialized by native Serdes.

By default, for values, deserialization on the inbound is natively performed by Kafka. Please note that this is a major change on default behavior from previous versions of Kafka Streams binder in which case the deserialization was done by the framework.

Kafka Streams binder will try to infer matching Serde types by looking at the type signature of java.util.function.Function|Consumer or StreamListener. Here is the order that it matches Serdes.

If none of the above strategies worked, then the applications must provide the Serdes through configuration. This can be configured in two ways - binding or default.

First the binder will look if a Serde is provided at the binding level. For e.g. if you have the following processor,

then, you can provide a binding level Serde using the following:

If you want the default key/value Serdes to be used for inbound deserialization, you can do so at the binder level.

If you don’t want the native decoding provided by Kafka, you can rely on the message conversion features that Spring Cloud Stream provides. Since native decoding is the default, in order to let Spring Cloud Stream deserialze the inbound value object, you need to explicitly disable native decoding.

For e.g. if you have the same BiFunction processor as above, then spring.cloud.stream.bindings.process_in_0.nativeDecoding: false You need to disable native decoding for all the inputs individually. Otherwise, native decoding will still be applied for those you don’t disable.

By default, Spring Cloud Stream will use application/json as the content type and use an appropriate json message converter. You can use custom message converters by using the following property.

Outbound serialization pretty much follows the same rules as above for inbound deserialization. As with the inbound deserialization, one major change from the previous versions of Spring Cloud Stream is that the serialization on the outbound is handled by Kafka natively. Before 3.0 versions of the binder, this was done by the framework itself.

Keys on the outbound are always serialized by Kafka using a matching Serde that is inferred by the binder. If it can’t infer the type of the key, then that needs to be specified using configuration.

Value serdes are inferred using the same rules used for inbound deserialization. First it matches to see if the outbound type is of a Serde exposed by Kafka such as - Long, Short, Double, Float, byte[] and String. If that doesnt’t work, then fall back to JsonSerde provided by the Spring Kafka project. Keep in mind that all these happen transparently to the application. If none of these work, then the user has to provide the Serde to use by configuration.

Lets say you are using the same BiFunction processor as above. Then you can configure outbound key/value Serdes as following.

However, falling back to default Serdes for both input deserialization and output serialization is the last resort. This may or may not work. Therefore, you need to ensure that you have a path forward for the application to correctly retrive the Serde.

If Serde inference fails, no binding level Serdes are provided, then the binder falls back to the default Serdes.

spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde

If your application uses the branching feature and has multiple output bindings, then these have to be configured per binding. Once again, if the binder is capable of inferring the Serde types, you don’t need to do this configuration.

If you don’t want the native encoding provided by Kafka, but want to use the framework provided message conversion, then you need to explicitly disable native decoding since since native decoding is the default. For e.g. if you have the same BiFunction processor as above, then spring.cloud.stream.bindings.process_out.nativeEncoding: false You need to disable native encoding for all the output individually in the case of branching. Otherwise, native encoding will still be applied for those you don’t disable.

By default, Spring Cloud Stream will use application/json as the content type and use an appropriate json message converter. You can use custom message converters by using the following property.

When native encoding/decoding is disabled, binder will not do any inference as in the case of native Serdes. Applications need to explicitly provide all the configuration options. For that reason, it is generally advised to stay with the default options for de/serialization and stick with native de/serialization provided by Kafka Streams when you write Spring Cloud Stream Kafka Streams applications. The one scenario in which you must use message conversion capabilities provided by the framework is when your upstream producer is using a specific serialization strategy. In that case, you want to use a matching deserialization strategy as native mechanisms may fail.

The following properties are available at the binder level and must be prefixed with spring.cloud.stream.kafka.streams.binder.

For more information about all the properties that may go into streams configuration, see StreamsConfig JavaDocs in Apache Kafka Streams docs.

The following properties are only available for Kafka Streams producers and must be prefixed with spring.cloud.stream.kafka.streams.bindings.<binding name>.producer. For convenience, if there multiple output bindings and they all require a common value, that can be configured by using the prefix spring.cloud.stream.kafka.streams.default.producer..

The following properties are available for Kafka Streams consumers and must be prefixed with spring.cloud.stream.kafka.streams.bindings.<binding-name>.consumer. For convenience, if there are multiple input bindings and they all require a common value, that can be configured by using the prefix spring.cloud.stream.kafka.streams.default.consumer..

Note: Using resetOffsets on the consumer does not have any effect on Kafka Streams binder. Unlike the message channel based binder, Kafka Streams binder does not seek to beginning or end on demand.

For general error handling in Kafka Streams binder, it is up to the end user applications to handle application level errors. As a side effect of providing a DLQ for deserialization exception handlers, Kafka Streams binder 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.

It continues to remain hard to robust error handling using the high-level DSL; Kafka Streams doesn’t natively support error handling yet.

However, when you use the low-level Processor API in your application, there are options to control this behavior. See below.

You can define custom state stores as beans in your application and those will be detected and added to the Kafka Streams builder by the binder. Note that, for regular StreamListener based processors, you still need to use the KafkaStreamsStateStore annotation for custom state stores. Here is an example of using custom state stores with functional style described in this section.

These state stores can be then accessed by the applications directly.

We recommend the m2eclipe eclipse plugin when working with eclipse. If you don’t already have m2eclipse installed it is available from the "eclipse marketplace".

Unfortunately m2e does not yet support Maven 3.3, so once the projects are imported into Eclipse you will also need to tell m2eclipse to use the .settings.xml file for the projects. If you do not do this you may see many different errors related to the POMs in the projects. Open your Eclipse preferences, expand the Maven preferences, and select User Settings. In the User Settings field click Browse and navigate to the Spring Cloud project you imported selecting the .settings.xml file in that project. Click Apply and then OK to save the preference changes.

If you prefer not to use m2eclipse you can generate eclipse project metadata using the following command:

The generated eclipse projects can be imported by selecting import existing projects from the file menu.

[[contributing] == Contributing

Spring Cloud is released under the non-restrictive Apache 2.0 license, and follows a very standard Github development process, using Github tracker for issues and merging pull requests into master. If you want to contribute even something trivial please do not hesitate, but follow the guidelines below.