SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix indentation for all pages

Browse Source
This commit is contained in:
Marcin Grzejszczak
2023-09-08 16:12:37 +02:00
parent f8217b2ad4
commit 271e573db4
14 changed files with 282 additions and 280 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
docs/modules/ROOT/pages/README.adoc
View File

@@ -11,20 +11,21 @@ image::https://codecov.io/gh/spring-cloud/spring-cloud-stream/branch/{github-tag
// ======================================================================================
[[introduction]]
== Introduction
= Introduction
include::intro.adoc[]
[[resources]]
== Resources
= Resources
For more information, please visit the https://spring.io/projects/spring-cloud-stream[project website]:
[[building]]
== Building
= Building
include::https://raw.githubusercontent.com/spring-cloud/spring-cloud-build/master/docs/src/main/asciidoc/building.adoc[]
[[contributing]]
== Contributing
= Contributing
include::https://raw.githubusercontent.com/spring-cloud/spring-cloud-build/master/docs/src/main/asciidoc/contributing.adoc[]

2
docs/modules/ROOT/pages/binders.adoc
View File

@@ -1,7 +1,7 @@
*{project-version}*
[[binders]]
== Binder Implementations
= Binder Implementations
The following binder implementations are available:

12
docs/modules/ROOT/pages/intro.adoc
View File

@@ -8,10 +8,10 @@ While these two binder implementations are based on Message Channels, Spring Clo
Below, you can find more information on how to use these various out-of-the-box binder implementations in your applications.
[[apache-kafka-binder]]
== Apache Kafka Binder
= Apache Kafka Binder
[[usage]]
=== Usage
== Usage
To use Apache Kafka binder, you need to add `spring-cloud-stream-binder-kafka` as a dependency to your Spring Cloud Stream application, as shown in the following example for Maven:
@@ -34,10 +34,10 @@ Alternatively, you can also use the Spring Cloud Stream Kafka Starter, as shown
----
[[apache-kafka-streams-binder]]
== Apache Kafka Streams Binder
= Apache Kafka Streams Binder
[[usage]]
=== Usage
== Usage
To use Apache Kafka Streams binder, you need to add `spring-cloud-stream-binder-kafka-streams` as a dependency to your Spring Cloud Stream application, as shown in the following example for Maven:
@@ -50,9 +50,9 @@ To use Apache Kafka Streams binder, you need to add `spring-cloud-stream-binder-
----
[[rabbitmq-binder]]
== RabbitMQ Binder
= RabbitMQ Binder
[[usage]]
=== Usage
== Usage
To use the RabbitMQ binder, you can add it to your Spring Cloud Stream application, by using the following Maven coordinates:

16
docs/modules/ROOT/pages/kafka/kafka-reactive.adoc
View File

@@ -1,12 +1,12 @@
[[reactive-kafka-binder]]
== Reactive Kafka Binder
= Reactive Kafka Binder
Kafka binder in Spring Cloud Stream provides a dedicated reactive binder based on the https://projectreactor.io/docs/kafka/release/reference/[Reactor Kafka] project.
This reactive Kafka binder enables full end-to-end reactive capabilities such as backpressure, reactive streams etc. in applications based on Apache Kafka.
When your Spring Cloud Stream Kafka application is written using reactive types (`Flux`, `Mono` etc.), it is recommended to use this reactive Kafka binder instead of the regular message channel based Kafka binder.
[[maven-coordinates]]
=== Maven Coordinates
== Maven Coordinates
Following are the maven coordinates for the reactive Kafka binder.
@@ -18,7 +18,7 @@ Following are the maven coordinates for the reactive Kafka binder.
```
[[basic-example-using-the-reactive-kafka-binder]]
=== Basic Example using the Reactive Kafka Binder
== Basic Example using the Reactive Kafka Binder
In this section, we show some basic code snippets for writing a reactive Kafka application using the reactive binder and details around them.
@@ -96,7 +96,7 @@ Given that Spring has no access to the contents of the inner flux, the applicati
This requires the use of native decoding (by specifying a `Deserializer` of the appropriate type in the configuration) to return record keys/values of the desired types.
[[consuming-records-in-the-raw-format]]
=== Consuming Records in the Raw Format
== Consuming Records in the Raw Format
In the above `upppercase` function, we are consuming the record as `Flux<String>` and then produce it as `Flux<String>`.
There might be occasions in which you need to receive the record in the original received format - the `ReceiverRecord`.
@@ -163,7 +163,7 @@ If `reactiveAutoCommit` is `false` (default), call `rec.receiverOffset().acknowl
Refer to the `reactor-kafka` documentation and javadocs for more information.
[[concurrency]]
=== Concurrency
== Concurrency
When using reactive functions with the reactive Kafka binder, if you set concurrency on the consumer binding, then the binder creates as many dedicated `KafkaReceiver` objects as provided by the concurrency value.
In other words, this creates multiple reactive streams with separate `Flux` implementations.
@@ -179,19 +179,19 @@ spring.cloud.stream.bindings.lowercase-in-0.consumer.concurrency=3
That will create three dedicated `KafkaReceiver` objects that generate three separate `Flux` implementations and then stream them to the handler method.
[[multiplex]]
=== Multiplex
== Multiplex
Starting with version 4.0.3, the common consumer property `multiplex` is now supported by the reactive binder, where a single binding can consume from multiple topics.
When `false` (default), a separate binding is created for each topic specified in a comma-delimited list in the common `destination` property.
[[destination-is-pattern]]
=== Destination is Pattern
== Destination is Pattern
Starting with version 4.0.3, the `destination-is-pattern` Kafka binding consumer property is now supported.
The receiver options are conigured with a regex `Pattern`, allowing the binding to consume from any topic that matches the pattern.
[[sender-result-channel]]
=== Sender Result Channel
== Sender Result Channel
Starting with version 4.0.3, you can configure the `resultMetadataChannel` to receive `SenderResult<?>` s to determine success/failure of sends.

112
docs/modules/ROOT/pages/kafka/kafka-streams.adoc
View File

@@ -1,9 +1,9 @@
[[kafka-streams-binder]]
== Kafka Streams Binder
= Kafka Streams Binder
[[usage]]
=== Usage
== Usage
For using the Kafka Streams binder, you just need to add it to your Spring Cloud Stream application, using the following maven coordinates:
@@ -20,7 +20,7 @@ A quick way to bootstrap a new project for Kafka Streams binder is to use http:/
image::{github-raw}/docs/src/main/asciidoc/images/spring-initializr-kafka-streams.png[width=800,scaledwidth="75%",align="center"]
[[overview]]
=== Overview
== Overview
Spring Cloud Stream includes a binder implementation designed explicitly for https://kafka.apache.org/documentation/streams/[Apache Kafka Streams] binding.
With this native integration, a Spring Cloud Stream "processor" application can directly use the
@@ -36,13 +36,13 @@ Alternatively, a Processor application with no outbound destination can be defin
In the following sections, we are going to look at the details of Spring Cloud Stream's integration with Kafka Streams.
[[programming-model]]
=== Programming Model
== Programming Model
When using the programming model provided by Kafka Streams binder, both the high-level https://docs.confluent.io/current/streams/developer-guide/dsl-api.html[Streams DSL] and a mix of both the higher level and the lower level https://docs.confluent.io/current/streams/developer-guide/processor-api.html[Processor-API] can be used as options.
When mixing both higher and lower level API's, this is usually achieved by invoking `transform` or `process` API methods on `KStream`.
[[functional-style]]
==== Functional Style
=== Functional Style
Starting with Spring Cloud Stream `3.0.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`.
@@ -161,7 +161,7 @@ In addition, you can also provide topic patterns as destinations if you want to
`spring.cloud.stream.bindings.process-in-0.destination=input.*`
[[multiple-input-bindings]]
===== Multiple Input Bindings
==== Multiple Input Bindings
Many non-trivial Kafka Streams applications often consume data from more than one topic through multiple bindings.
For instance, one topic is consumed as `Kstream` and another as `KTable` or `GlobalKTable`.
@@ -171,7 +171,7 @@ If the application specifies that the data needs to be bound as `KTable` or `Glo
We will look at a few different scenarios how multiple input bindings are handled in the Kafka Streams binder.
[[bifunction-in-kafka-streams-binder]]
====== BiFunction in Kafka Streams Binder
===== BiFunction in Kafka Streams Binder
Here is an example where we have two inputs and an output. In this case, the application can leverage on `java.util.function.BiFunction`.
@@ -197,7 +197,7 @@ The default binding names generated by the binder for the inputs are `process-in
In this example, the first parameter of `BiFunction` is bound as a `KStream` for the first input and the second parameter is bound as a `KTable` for the second input.
[[biconsumer-in-kafka-streams-binder]]
====== BiConsumer in Kafka Streams Binder
===== BiConsumer in Kafka Streams Binder
If there are two inputs, but no outputs, in that case we can use `java.util.function.BiConsumer` as shown below.
@@ -210,7 +210,7 @@ public BiConsumer<KStream<String, Long>, KTable<String, String>> process() {
----
[[beyond-two-inputs]]
====== Beyond two inputs
===== Beyond two inputs
What if you have more than two inputs?
There are situations in which you need more than two inputs. In that case, the binder allows you to chain partial functions.
@@ -264,7 +264,7 @@ With curried functions, you can virtually have any number of inputs. However, ke
Therefore if your Kafka Streams application requires more than a reasonably smaller number of input bindings, and you want to use this functional model, then you may want to rethink your design and decompose the application appropriately.
[[output-bindings]]
===== Output Bindings
==== Output Bindings
Kafka Streams binder allows types of either `KStream` or `KTable` as output bindings.
Behind the scenes, the binder uses the `to` method on `KStream` to send the resultant records to the output topic.
@@ -286,7 +286,7 @@ public Function<KTable<String, String>, KStream<String, String>> bar() {
```
[[multiple-output-bindings]]
===== Multiple Output Bindings
==== Multiple Output Bindings
Kafka Streams allows writing outbound data into multiple topics. This feature is known as branching in Kafka Streams.
When using multiple output bindings, you need to provide an array of KStream (`KStream[]`) as the outbound return type.
@@ -328,7 +328,7 @@ The reason why the binder generates three output bindings is because it detects
Note that in this example, we provide a `noDefaultBranch()`; if we have used `defaultBranch()` instead, that would have required an extra output binding, essentially returning a `KStream` array of length four.
[[summary-of-function-based-programming-styles-for-kafka-streams]]
===== Summary of Function based Programming Styles for Kafka Streams
==== Summary of Function based Programming Styles for Kafka Streams
In summary, the following table shows the various options that can be used in the functional paradigm.
@@ -346,7 +346,7 @@ In summary, the following table shows the various options that can be used in th
* In the case of more than one output in this table, the type simply becomes `KStream[]`.
[[function-composition-in-kafka-streams-binder]]
===== Function composition in Kafka Streams binder
==== Function composition in Kafka Streams binder
Kafka Streams binder supports minimal forms of functional composition for linear topologies.
Using the Java functional API support, you can write multiple functions and then compose them on your own using the `andThen` method.
@@ -391,7 +391,7 @@ spring.cloud.function.definition=foo|bar;foo;bar
The composed function's default binding names in this example becomes `foobar-in-0` and `foobar-out-0`.
[[limitations-of-functional-composition-in-kafka-streams-bincer]]
====== Limitations of functional composition in Kafka Streams bincer
===== Limitations of functional composition in Kafka Streams bincer
When you have `java.util.function.Function` bean, that can be composed with another function or multiple functions.
The same function bean can be composed with a `java.util.function.Consumer` as well. In this case, consumer is the last component composed.
@@ -426,7 +426,7 @@ The default input bindings in this case are going to be `curriedFoobar-in-0` and
The default output binding for this example becomes `curriedFoobar-out-0`.
[[special-note-on-using-ktable-as-output-in-function-composition]]
====== Special note on using `KTable` as output in function composition
===== Special note on using `KTable` as output in function composition
Lets say you have the following two functions.
@@ -446,10 +446,10 @@ public Function<KTable<String, String>, KStream<String, String>> bar() {
You can compose them as `foo|bar`, but keep in mind that the second function (`bar` in this case) must have a `KTable` as input since the first function (`foo`) has `KTable` as output.
[[ancillaries-to-the-programming-model]]
=== Ancillaries to the programming model
== Ancillaries to the programming model
[[multiple-kafka-streams-processors-within-a-single-application]]
==== Multiple Kafka Streams processors within a single application
=== Multiple Kafka Streams processors within a single application
Binder allows to have multiple Kafka Streams processors within a single Spring Cloud Stream application.
You can have an application as below.
@@ -483,7 +483,7 @@ If you want certain functions to be not activated right away, you can remove tha
This is also true when you have a single Kafka Streams processor and other types of `Function` beans in the same application that is handled through a different binder (for e.g., a function bean that is based on the regular Kafka Message Channel binder)
[[kafka-streams-application-id]]
==== Kafka Streams Application ID
=== Kafka Streams Application ID
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.
@@ -535,14 +535,14 @@ The generated application ID in this manner will be static over application rest
In the case of functional model, the generated application ID will be the function bean name followed by the literal `applicationID`, for e.g `process-applicationID` if `process` if the function bean name.
[[summary-of-setting-application-id]]
====== Summary of setting Application ID
===== Summary of setting Application ID
* By default, binder will auto generate the application ID per function methods.
* If you have a single processor, then you can use `spring.kafka.streams.applicationId`, `spring.application.name` or `spring.cloud.stream.kafka.streams.binder.applicationId`.
* If you have multiple processors, then application ID can be set per function using the property - `spring.cloud.stream.kafka.streams.binder.functions.<function-name>.applicationId`.
[[overriding-the-default-binding-names-generated-by-the-binder-with-the-functional-style]]
==== Overriding the default binding names generated by the binder with the functional style
=== Overriding the default binding names generated by the binder with the functional style
By default, the binder uses the strategy discussed 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-0, process-out-0 etc.
If you want to override those binding names, you can do that by specifying the following properties.
@@ -576,7 +576,7 @@ Please keep in mind that with the functional programming model described above,
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.
[[setting-up-bootstrap-server-configuration]]
==== Setting up bootstrap server configuration
=== Setting up bootstrap server configuration
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`.
@@ -589,14 +589,14 @@ When it comes to the binder level property, it doesn't matter if you use the bro
Kafka Streams binder will first check if Kafka Streams binder specific broker property is set (`spring.cloud.stream.kafka.streams.binder.brokers`) and if not found, it looks for `spring.cloud.stream.kafka.binder.brokers`.
[[record-serialization-and-deserialization]]
=== Record serialization and deserialization
== Record serialization and deserialization
Kafka Streams binder allows you to serialize and deserialize records in two ways.
One is the native serialization and deserialization facilities provided by Kafka and the other one is the message conversion capabilities of Spring Cloud Stream framework.
Lets look at some details.
[[inbound-deserialization]]
==== Inbound deserialization
=== Inbound deserialization
Keys are always deserialized using native Serdes.
@@ -674,7 +674,7 @@ spring.cloud.stream.bindings.process-in-0.contentType
```
[[outbound-serialization]]
==== Outbound serialization
=== Outbound serialization
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.
@@ -729,7 +729,7 @@ Your business logic might still need to call Kafka Streams API's that explicitly
Those are still the responsibility of the application and must be handled accordingly by the developer.
[[error-handling]]
=== Error Handling
== Error Handling
Apache Kafka Streams provides 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].
@@ -737,7 +737,7 @@ Out of the box, Apache Kafka Streams provides two kinds of deserialization excep
As the name indicates, the former will log the error and continue processing the next records and the latter will log the error and fail. `LogAndFailExceptionHandler` is the default deserialization exception handler.
[[handling-deserialization-exceptions-in-the-binder]]
==== Handling Deserialization Exceptions in the Binder
=== Handling Deserialization Exceptions in the Binder
Kafka Streams binder allows to specify the deserialization exception handlers above using the following property.
@@ -804,7 +804,7 @@ For instance, if your binding's destination topic is `inputTopic` and the applic
It is always recommended to explicitly create a DLQ topic for each input binding if it is your intention to enable DLQ.
[[dlq-per-input-consumer-binding]]
==== DLQ per input consumer binding
=== DLQ per input consumer binding
The property `spring.cloud.stream.kafka.streams.binder.deserializationExceptionHandler` is applicable for the entire application.
This implies that if there are multiple functions in the same application, this property is applied to all of them.
@@ -827,7 +827,7 @@ and you only want to enable DLQ on the first input binding and skipAndContinue o
Setting deserialization exception handlers this way has a higher precedence than setting at the binder level.
[[dlq-partitioning]]
==== DLQ partitioning
=== DLQ partitioning
By default, records are published to the Dead-Letter topic using the same partition as the original record.
This means the Dead-Letter topic must have at least as many partitions as the original record.
@@ -857,13 +857,13 @@ This implies that if there are multiple functions in the same application, this
* The exception handling for deserialization works consistently with native deserialization and framework provided message conversion.
[[handling-production-exceptions-in-the-binder]]
==== Handling Production Exceptions in the Binder
=== Handling Production Exceptions in the Binder
Unlike the support for deserialization exception handlers as described above, the binder does not provide such first class mechanisms for handling production exceptions.
However, you still can configure production exception handlers using the `StreamsBuilderFactoryBean` customizer which you can find more details about, in a subsequent section below.
[[runtime-error-handling]]
==== Runtime Error Handling
=== Runtime Error Handling
When it comes to handling errors from application code, i.e. from the business logic execution, it is usually up to the application to handle that.
Because, the Kafka Streams binder does not have a way to interfere with the application code.
@@ -924,7 +924,7 @@ In this case, when the record fails, the `DltAwareProcessor`, instead of using i
In this case also, you can provide an optional `Supplier<Long>` to dictate the timestamp used in the record passed in to the `BiConsumer` recoverer.
[[retrying-critical-business-logic]]
=== Retrying critical business logic
== Retrying critical business logic
There are scenarios in which you might want to retry parts of your business logic that are critical to the application.
There maybe an external call to a relational database or invoking a REST endpoint from the Kafka Streams processor.
@@ -1029,7 +1029,7 @@ retryTemplate.execute(context -> {
Refer to the https://github.com/spring-projects/spring-retry[Spring Retry] project for more information about the RetryTemplate, retry policies, backoff policies and more.
[[state-store]]
=== State Store
== State Store
State stores are created automatically by Kafka Streams when the high level DSL is used and appropriate calls are made those trigger a state store.
@@ -1098,7 +1098,7 @@ This will not work when it comes to registering global state stores.
In order to register a global state store, please see the section below on customizing `StreamsBuilderFactoryBean`.
[[interactive-queries]]
=== Interactive Queries
== Interactive Queries
Kafka Streams binder API exposes a class called `InteractiveQueryService` to interactively query the state stores.
You can access this as a Spring bean in your application. An easy way to get access to this bean from your application is to `autowire` the bean.
@@ -1158,7 +1158,7 @@ For these methods also, during startup, if the underlying KafkaStreams objects a
The aforementioned retry properties are applicable for these methods as well.
[[other-api-methods-available-through-the-interactivequeryservice]]
==== Other API methods available through the InteractiveQueryService
=== Other API methods available through the InteractiveQueryService
Use the following API method to retrieve the `KeyQueryMetadata` object associated with the combination of given store and key.
@@ -1173,7 +1173,7 @@ public <K> KafkaStreams getKafkaStreams(String store, K key, Serializer<K> seria
```
[[customizing-store-query-parameters]]
==== Customizing Store Query Parameters
=== Customizing Store Query Parameters
Sometimes it is necessary that you need to fine tune the store query parameters before querying the store through `InteractiveQueryService`.
For this purpose, starting with the `4.0.1` version of the binder, you can provide a bean for `StoreQueryParametersCustomizer` which is a functional interface with a `customize` method that takes a `StoreQueryParameter` as the argument.
@@ -1190,7 +1190,7 @@ When this bean is present in this application, `InteractiveQueryService` will ca
NOTE: Keep in mind that, there must be a unique bean for `StoreQueryParametersCustomizer` available in the application.
[[health-indicator]]
=== Health Indicator
== Health Indicator
The health indicator requires the dependency `spring-boot-starter-actuator`. For maven use:
[source,xml]
@@ -1224,7 +1224,7 @@ When enabling `show-details`, some of the information reported may be redundant.
When there are multiple Kafka Streams processors present in the same application, then the health checks will be reported for all of them and will be categorized by the application ID of Kafka Streams.
[[accessing-kafka-streams-metrics]]
=== Accessing Kafka Streams Metrics
== Accessing Kafka Streams Metrics
Spring Cloud Stream Kafka Streams binder provides Kafka Streams metrics which can be exported through a Micrometer `MeterRegistry`.
@@ -1235,7 +1235,7 @@ When accessing metrics through the Boot actuator endpoint, make sure to add `met
Then you can access `/acutator/metrics` to get a list of all the available metrics, which then can be individually accessed through the same URI (`/actuator/metrics/<metric-name>`).
[[mixing-high-level-dsl-and-low-level-processor-api]]
=== Mixing high level DSL and low level Processor API
== Mixing high level DSL and low level Processor API
Kafka Streams provides two variants of APIs.
It has a higher level DSL like API where you can chain various operations that maybe familiar to a lot of functional programmers.
@@ -1298,7 +1298,7 @@ public Consumer<KStream<Object, String>> process() {
The `process` API method call is a terminal operation while the `transform` API is non terminal and gives you a potentially transformed `KStream` using which you can continue further processing using either the DSL or the processor API.
[[partition-support-on-the-outbound]]
=== Partition support on the outbound
== Partition support on the outbound
A Kafka Streams processor usually sends the processed output into an outbound Kafka topic.
If the outbound topic is partitioned and the processor needs to send the outgoing data into particular partitions, the applications needs to provide a bean of type `StreamPartitioner`.
@@ -1357,7 +1357,7 @@ spring.cloud.stream.kafka.streams.bindings.process-out-0.producer.streamPartitio
Each output topic in the application needs to be configured separately like this.
[[streamsbuilderfactorybean-customizer]]
=== StreamsBuilderFactoryBean customizer
== StreamsBuilderFactoryBean customizer
It is often required to customize the `StreamsBuilderFactoryBean` that creates the `KafkaStreams` objects.
Based on the underlying support provided by Spring Kafka, the binder allows you to customize the `StreamsBuilderFactoryBean`.
@@ -1427,7 +1427,7 @@ public StreamsBuilderFactoryBeanCustomizer streamsBuilderFactoryBeanCustomizer()
```
[[using-customizer-to-register-a-global-state-store]]
==== Using Customizer to register a global state store
=== Using Customizer to register a global state store
As mentioned above, the binder does not provide a first class way to register global state stores as a feature.
For that, you need to use the customizer.
@@ -1451,7 +1451,7 @@ public StreamsBuilderFactoryBeanCustomizer customizer() {
Again, if you have multiple processors, you want to attach the global state store to the right `StreamsBuilder` by filtering out the other `StreamsBuilderFactoryBean` objects using the application id as outlined above.
[[using-customizer-to-register-a-production-exception-handler]]
==== Using customizer to register a production exception handler
=== Using customizer to register a production exception handler
In the error handling section, we indicated that the binder does not provide a first class way to deal with production exceptions.
Though that is the case, you can still use the `StreamsBuilderFacotryBean` customizer to register production exception handlers. See below.
@@ -1470,7 +1470,7 @@ Once again, if you have multiple processors, you may want to set it appropriatel
You may also add such production exception handlers using the configuration property (See below for more on that), but this is an option if you choose to go with a programmatic approach.
[[timestamp-extractor]]
=== Timestamp extractor
== Timestamp extractor
Kafka Streams allows you to control the processing of the consumer records based on various notions of timestamp.
By default, Kafka Streams extracts the timestamp metadata embedded in the consumer record.
@@ -1504,7 +1504,7 @@ spring.cloud.stream.kafka.streams.bindings.process-in-2.consumer.timestampExtrac
If you skip an input consumer binding for setting a custom timestamp extractor, that consumer will use the default settings.
[[multi-binders-with-kafka-streams-based-binders-and-regular-kafka-binder]]
=== Multi binders with Kafka Streams based binders and regular Kafka Binder
== Multi binders with Kafka Streams based binders and regular Kafka Binder
You can have an application where you have both a function/consumer/supplier that is based on the regular Kafka binder and a Kafka Streams based processor.
However, you cannot mix both of them within a single function or consumer.
@@ -1605,7 +1605,7 @@ spring.cloud.stream.bindings.enrichOrder-out-0.binder=kafka1 #kstream
```
[[state-cleanup]]
=== State Cleanup
== State Cleanup
By default, no local state is cleaned up when the binding is stopped.
This is the same behavior effective from Spring Kafka version 2.7.
@@ -1613,7 +1613,7 @@ See https://docs.spring.io/spring-kafka/reference/html/#streams-config[Spring Ka
To modify this behavior simply add a single `CleanupConfig` `@Bean` (configured to clean up on start, stop, or neither) to the application context; the bean will be detected and wired into the factory bean.
[[kafka-streams-topology-visualization]]
=== Kafka Streams topology visualization
== Kafka Streams topology visualization
Kafka Streams binder provides the following actuator endpoints for retrieving the topology description using which you can visualize the topology using external tools.
@@ -1626,7 +1626,7 @@ Further, you also need to add `kafkastreamstopology` to `management.endpoints.we
By default, the `kafkastreamstopology` endpoint is disabled.
[[event-type-based-routing-in-kafka-streams-applications]]
=== Event type based routing in Kafka Streams applications
== Event type based routing in Kafka Streams applications
Routing functions available in regular message channel based binders are not supported in Kafka Streams binder.
However, Kafka Streams binder still provides routing capabilities through the event type record header on the inbound records.
@@ -1669,7 +1669,7 @@ In order to address this issue, you can set the following property on the bindin
This way, the application can detect deserialization issues right away when using the event routing feature and can take appropriate handling decisions.
[[binding-visualization-and-control-in-kafka-streams-binder]]
=== Binding visualization and control in Kafka Streams binder
== Binding visualization and control in Kafka Streams binder
Starting with version 3.1.2, Kafka Streams binder supports binding visualization and control.
The only two lifecycle phases supported are `STOPPED` and `STARTED`.
@@ -1782,7 +1782,7 @@ This is because all the bindings on a single function are backed by the same `St
Therefore, for the function above, either `function-in-0` or `function-out-0` will work.
[[manually-starting-kafka-streams-processors]]
=== Manually starting Kafka Streams processors
== Manually starting Kafka Streams processors
Spring Cloud Stream Kafka Streams binder offers an abstraction called `StreamsBuilderFactoryManager` on top of the `StreamsBuilderFactoryBean` from Spring for Apache Kafka.
This manager API is used for controlling the multiple `StreamsBuilderFactoryBean` per processor in a binder based application.
@@ -1807,7 +1807,7 @@ In order to correct this, you can set `spring.kafka.streams.auto-startup` to `fa
Keep in mind that, when using the Spring Cloud Stream binder, you are not directly dealing with `StreamsBuilderFactoryBean` from Spring for Apache Kafka, rather `StreamsBuilderFactoryManager`, as the `StreamsBuilderFactoryBean` objects are internally managed by the binder.
[[manually-starting-kafka-streams-processors-selectively]]
=== Manually starting Kafka Streams processors selectively
== Manually starting Kafka Streams processors selectively
While the approach laid out above will unconditionally apply auto start `false` to all the Kafka Streams processors in the application through `StreamsBuilderFactoryManager`, it is often desirable that only individually selected Kafka Streams processors are not auto started.
For instance, let us assume that you have three different functions (processors) in your application and for one of the processors, you do not want to start it as part of the application startup.
@@ -1880,7 +1880,7 @@ NOTE: When controlling the bindings by disabling `auto-startup` as described in
In other words, if you use the producer binding, `process3-out-0`, that does not have any effect in terms of disabling the auto starting of the processor, although this producer binding uses the same `StreamsBuilderFactoryBean` as the consumer bindings.
[[tracing-using-spring-cloud-sleuth]]
=== Tracing using Spring Cloud Sleuth
== Tracing using Spring Cloud Sleuth
When Spring Cloud Sleuth is on the classpath of a Spring Cloud Stream Kafka Streams binder based application, both its consumer and producer are automatically instrumented with tracing information.
However, in order to trace any application specific operations, those need to be explicitly instrumented by the user code.
@@ -1944,14 +1944,14 @@ public Function<KStream<String, String>, KStream<String, String>> process(KafkaS
```
[[configuration-options]]
=== Configuration Options
== Configuration Options
This section contains the configuration options used by the Kafka Streams binder.
For common configuration options and properties pertaining to binder, refer to the <<binding-properties,core documentation>>.
[[kafka-streams-binder-properties]]
==== Kafka Streams Binder Properties
=== Kafka Streams Binder Properties
The following properties are available at the binder level and must be prefixed with `spring.cloud.stream.kafka.streams.binder.`
Any Kafka binder provided properties re-used in Kafka Streams binder must be prefixed with `spring.cloud.stream.kafka.streams.binder` instead of `spring.cloud.stream.kafka.binder`.
@@ -2034,7 +2034,7 @@ Set this property to `true` to enable health check for all processors including
Default: false
[[kafka-streams-producer-properties]]
==== Kafka Streams Producer Properties
=== Kafka Streams Producer Properties
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 are 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.`.
@@ -2066,7 +2066,7 @@ Custom name for the sink component to which the processor is producing to.
Deafult: `none` (generated by Kafka Streams)
[[kafka-streams-consumer-properties]]
==== Kafka Streams Consumer Properties
=== Kafka Streams Consumer Properties
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.`.
@@ -2141,7 +2141,7 @@ Custom name for the source component from which the processor is consuming from.
Deafult: `none` (generated by Kafka Streams)
[[special-note-on-concurrency]]
==== Special note on concurrency
=== Special note on concurrency
In Kafka Streams, you can control of the number of threads a processor can create using the `num.stream.threads` property.
This, you can do using the various `configuration` options described above under binder, functions, producer or consumer level.

6
docs/modules/ROOT/pages/kafka/kafka_dlq.adoc
View File

@@ -1,8 +1,8 @@
[[kafka-dlq-processing]]
=== Dead-Letter Topic Processing
= Dead-Letter Topic Processing
[[dlq-partition-selection]]
==== Dead-Letter Topic Partition Selection
== Dead-Letter Topic Partition Selection
By default, records are published to the Dead-Letter topic using the same partition as the original record.
This means the Dead-Letter topic must have at least as many partitions as the original record.
@@ -52,7 +52,7 @@ This is because there is no way for the binder to infer the names of all the DLQ
Therefore, if you provide DLQ names using this strategy, it is the application's responsibility to ensure that those topics are created beforehand.
[[dlq-handling]]
==== Handling Records in a Dead-Letter Topic
== Handling Records in a Dead-Letter Topic
Because the framework cannot anticipate how users would want to dispose of dead-lettered messages, it does not provide any standard mechanism to handle them.
If the reason for the dead-lettering is transient, you may wish to route the messages back to the original topic.

52
docs/modules/ROOT/pages/kafka/kafka_overview.adoc
View File

@@ -6,10 +6,10 @@ In addition, this guide explains the Kafka Streams binding capabilities of Sprin
--
[[apache-kafka-binder]]
== Apache Kafka Binder
= Apache Kafka Binder
[[usage]]
=== Usage
== Usage
To use Apache Kafka binder, you need to add `spring-cloud-stream-binder-kafka` as a dependency to your Spring Cloud Stream application, as shown in the following example for Maven:
@@ -32,7 +32,7 @@ Alternatively, you can also use the Spring Cloud Stream Kafka Starter, as shown
----
[[overview]]
=== Overview
== Overview
The following image shows a simplified diagram of how the Apache Kafka binder operates:
@@ -49,14 +49,14 @@ For example, with versions earlier than 0.11.x.x, native headers are not support
Also, 0.11.x.x does not support the `autoAddPartitions` property.
[[configuration-options]]
=== Configuration Options
== Configuration Options
This section contains the configuration options used by the Apache Kafka binder.
For common configuration options and properties pertaining to the binder, see the https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/current/reference/html/spring-cloud-stream.html#binding-properties[binding properties] in core documentation.
[[kafka-binder-properties]]
==== Kafka Binder Properties
=== Kafka Binder Properties
spring.cloud.stream.kafka.binder.brokers::
A list of brokers to which the Kafka binder connects.
@@ -185,7 +185,7 @@ Enable Micrometer observation registry on all the bindings in this binder.
Default: false
[[kafka-consumer-properties]]
==== Kafka Consumer Properties
=== Kafka Consumer Properties
NOTE: To avoid repetition, Spring Cloud Stream supports setting values for all channels, in the format of `spring.cloud.stream.kafka.default.consumer.<property>=<value>`.
@@ -357,7 +357,7 @@ Default: none.
[[reset-offsets]]
==== Resetting Offsets
=== Resetting Offsets
When an application starts, the initial position in each assigned partition depends on two properties `startOffset` and `resetOffsets`.
If `resetOffsets` is `false`, normal Kafka consumer https://kafka.apache.org/documentation/#consumerconfigs_auto.offset.reset[`auto.offset.reset`] semantics apply.
@@ -384,7 +384,7 @@ IMPORTANT: If a rebalance occurs after the initial assignment, the seeks will on
For more control over topic offsets, see <<rebalance-listener>>; when a listener is provided, `resetOffsets` should not be set to `true`, otherwise, that will cause an error.
[[consuming-batches]]
==== Consuming Batches
=== Consuming Batches
Starting with version 3.0, when `spring.cloud.stream.bindings.<name>.consumer.batch-mode` is set to `true`, all of the records received by polling the Kafka `Consumer` will be presented as a `List<?>` to the listener method.
Otherwise, the method will be called with one record at a time.
@@ -399,7 +399,7 @@ You can also use a manual `AckMode` and call `Ackowledgment.nack(index, sleep)`
Refer to the https://docs.spring.io/spring-kafka/docs/2.3.15.BUILD-SNAPSHOT/reference/html/#committing-offsets[Spring for Apache Kafka documentation] for more information about these techniques.
[[kafka-producer-properties]]
==== Kafka Producer Properties
=== Kafka Producer Properties
NOTE: To avoid repetition, Spring Cloud Stream supports setting values for all channels, in the format of `spring.cloud.stream.kafka.default.producer.<property>=<value>`.
@@ -523,12 +523,12 @@ If set to true, records published to this output binding will not be run in a tr
Default: `false`
[[usage-examples]]
==== Usage examples
=== Usage examples
In this section, we show the use of the preceding properties for specific scenarios.
[[example:-setting-ackmode-to-manual-and-relying-on-manual-acknowledgement]]
===== Example: Setting `ackMode` to `MANUAL` and Relying on Manual Acknowledgement
==== Example: Setting `ackMode` to `MANUAL` and Relying on Manual Acknowledgement
This example illustrates how one may manually acknowledge offsets in a consumer application.
@@ -557,7 +557,7 @@ public class ManuallyAcknowdledgingConsumer {
----
[[example:-security-configuration]]
===== Example: Security Configuration
==== Example: Security Configuration
Apache Kafka 0.9 supports secure connections between client and brokers.
To take advantage of this feature, follow the guidelines in the https://kafka.apache.org/090/documentation.html#security_configclients[Apache Kafka Documentation] as well as the Kafka 0.9 https://docs.confluent.io/2.0.0/kafka/security.html[security guidelines from the Confluent documentation].
@@ -577,7 +577,7 @@ When using Kerberos, follow the instructions in the https://kafka.apache.org/090
Spring Cloud Stream supports passing JAAS configuration information to the application by using a JAAS configuration file and using Spring Boot properties.
[[using-jaas-configuration-files]]
====== Using JAAS Configuration Files
===== Using JAAS Configuration Files
The JAAS and (optionally) krb5 file locations can be set for Spring Cloud Stream applications by using system properties.
The following example shows how to launch a Spring Cloud Stream application with SASL and Kerberos by using a JAAS configuration file:
@@ -591,7 +591,7 @@ The following example shows how to launch a Spring Cloud Stream application with
----
[[using-spring-boot-properties]]
====== Using Spring Boot Properties
===== Using Spring Boot Properties
As an alternative to having a JAAS configuration file, Spring Cloud Stream provides a mechanism for setting up the JAAS configuration for Spring Cloud Stream applications by using Spring Boot properties.
@@ -648,7 +648,7 @@ Consequently, relying on Spring Cloud Stream to create/modify topics may fail.
In secure environments, we strongly recommend creating topics and managing ACLs administratively by using Kafka tooling.
[[multi-binder-configuration-and-jaas]]
====== Multi-binder configuration and JAAS
===== Multi-binder configuration and JAAS
When connecting to multiple clusters in which each one requires separate JAAS configuration, then set the JAAS configuration using the property `sasl.jaas.config`.
When this property is present in the application, it takes precedence over the other strategies mentioned above.
@@ -690,7 +690,7 @@ Note that both the Kafka clusters, and the `sasl.jaas.config` values for each of
See this https://github.com/spring-cloud/spring-cloud-stream-samples/tree/main/multi-binder-samples/kafka-multi-binder-jaas[sample application] for more details on how to setup and run such an application.
[[pause-resume]]
===== Example: Pausing and Resuming the Consumer
==== Example: Pausing and Resuming the Consumer
If you wish to suspend consumption but not cause a partition rebalance, you can pause and resume the consumer.
This is facilitated by managing the binding lifecycle as shown in **Binding visualization and control** in the Spring Cloud Stream documentation, using `State.PAUSED` and `State.RESUMED`.
@@ -699,7 +699,7 @@ To resume, you can use an `ApplicationListener` (or `@EventListener` method) to
The frequency at which events are published is controlled by the `idleEventInterval` property.
[[kafka-transactional-binder]]
=== Transactional Binder
== Transactional Binder
Enable transactions by setting `spring.cloud.stream.kafka.binder.transaction.transactionIdPrefix` to a non-empty value, e.g. `tx-`.
When used in a processor application, the consumer starts the transaction; any records sent on the consumer thread participate in the same transaction.
@@ -753,7 +753,7 @@ If you wish to synchronize producer-only transactions with those from some other
IMPORTANT: If you deploy multiple instances of your application, each instance needs a unique `transactionIdPrefix`.
[[kafka-error-channels]]
=== Error Channels
== Error Channels
Starting with version 1.3, the binder unconditionally sends exceptions to an error channel for each consumer destination and can also be configured to send async producer send failures to an error channel.
See https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/current/reference/html/spring-cloud-stream.html#spring-cloud-stream-overview-error-handling[this section on error handling] for more information.
@@ -767,7 +767,7 @@ There is no automatic handling of producer exceptions (such as sending to a <<ka
You can consume these exceptions with your own Spring Integration flow.
[[kafka-metrics]]
=== Kafka Metrics
== Kafka Metrics
Kafka binder module exposes the following metrics:
@@ -795,7 +795,7 @@ class NoOpBindingMeters {
More details on how to suppress meters selectively can be found https://micrometer.io/docs/concepts#_meter_filters[here].
[[kafka-tombstones]]
=== Tombstone Records (null record values)
== Tombstone Records (null record values)
When using compacted topics, a record with a `null` value (also called a tombstone record) represents the deletion of a key.
To receive such messages in a Spring Cloud Stream function, you can use the following strategy.
@@ -820,7 +820,7 @@ public Function<Message<Person>, String> myFunction() {
====
[[rebalance-listener]]
=== Using a KafkaBindingRebalanceListener
== Using a KafkaBindingRebalanceListener
Applications may wish to seek topics/partitions to arbitrary offsets when the partitions are initially assigned, or perform other operations on the consumer.
Starting with version 2.1, if you provide a single `KafkaBindingRebalanceListener` bean in the application context, it will be wired into all Kafka consumer bindings.
@@ -871,7 +871,7 @@ public interface KafkaBindingRebalanceListener {
You cannot set the `resetOffsets` consumer property to `true` when you provide a rebalance listener.
[[retry-and-dlq-processing]]
=== Retry and Dead Letter Processing
== Retry and Dead Letter Processing
By default, when you configure retry (e.g. `maxAttemts`) and `enableDlq` in a consumer binding, these functions are performed within the binder, with no participation by the listener container or Kafka consumer.
@@ -974,7 +974,7 @@ public BinderCustomizer binderCustomizer(ListenerContainerWithDlqAndRetryCustomi
====
[[consumer-producer-config-customizer]]
=== Customizing Consumer and Producer configuration
== Customizing Consumer and Producer configuration
If you want advanced customization of consumer and producer configuration that is used for creating `ConsumerFactory` and `ProducerFactory` in Kafka,
you can implement the following customizers.
@@ -989,7 +989,7 @@ When the binder discovers that these customizers are available as beans, it will
Both of these interfaces also provide access to both the binding and destination names so that they can be accessed while customizing producer and consumer properties.
[[admin-client-config-customization]]
=== Customizing AdminClient Configuration
== Customizing AdminClient Configuration
As with consumer and producer config customization above, applications can also customize the configuration for admin clients by providing an `AdminClientConfigCustomizer`.
AdminClientConfigCustomizer's configure method provides access to the admin client properties, using which you can define further customization.
@@ -1006,7 +1006,7 @@ public AdminClientConfigCustomizer adminClientConfigCustomizer() {
```
[[custom-kafka-binder-health-indicator]]
=== Custom Kafka Binder Health Indicator
== Custom Kafka Binder Health Indicator
Kafka binder activates a default health indicator when Spring Boot actuator is on the classpath.
This health indicator checks the health of the binder and any communication issues with the Kafka broker.
@@ -1030,7 +1030,7 @@ public KafkaBinderHealth kafkaBinderHealthIndicator() {
```
[[custom-kafka-binder-health-indicator-example]]
=== Custom kafka Binder Health Indicator Example
== Custom kafka Binder Health Indicator Example
Here is the pseudo-code for writing a custom Kafka binder HealthIndicator.
In this example, we try to override the binder provided Kafka HealthIndicator by specifically checking first for cluster connectivity and then followed by topic-related issues.

2
docs/modules/ROOT/pages/kafka/kafka_partitions.adoc
View File

@@ -1,5 +1,5 @@
[[partitioning-with-the-kafka-binder]]
=== Partitioning with the Kafka Binder
= Partitioning with the Kafka Binder
Apache Kafka supports topic partitioning natively.

92
docs/modules/ROOT/pages/kafka/kafka_tips.adoc
View File

@@ -1,18 +1,18 @@
[[tips-tricks-and-recipes]]
== Tips, Tricks and Recipes
= Tips, Tricks and Recipes
[[simple-dlq-with-kafka]]
=== Simple DLQ with Kafka
== Simple DLQ with Kafka
[[problem-statement]]
==== Problem Statement
=== Problem Statement
As a developer, I want to write a consumer application that processes records from a Kafka topic.
However, if some error occurs in processing, I don't want the application to stop completely.
Instead, I want to send the record in error to a DLT (Dead-Letter-Topic) and then continue processing new records.
[[solution]]
==== Solution
=== Solution
The solution for this problem is to use the DLQ feature in Spring Cloud Stream.
For the purposes of this discussion, let us assume that the following is our processor function.
@@ -85,15 +85,15 @@ spring.cloud.stream:
```
[[dlq-with-advanced-retry-options]]
=== DLQ with Advanced Retry Options
== DLQ with Advanced Retry Options
[[problem-statement]]
==== Problem Statement
=== Problem Statement
This is similar to the recipe above, but as a developer I would like to configure the way retries are handled.
[[solution]]
==== Solution
=== Solution
If you followed the above recipe, then you get the default retry options built into the Kafka binder when the processing encounters an error.
@@ -132,17 +132,17 @@ spring.cloud.stream.bindings.processData-in-0.consumer.retry-template-name=<your
```
[[handling-deserialization-errors-with-dlq]]
=== Handling Deserialization errors with DLQ
== Handling Deserialization errors with DLQ
[[problem-statement]]
==== Problem Statement
=== Problem Statement
I have a processor that encounters a deserialization exception in Kafka consumer.
I would expect that the Spring Cloud Stream DLQ mechanism will catch that scenario, but it does not.
How can I handle this?
[[solution]]
==== Solution
=== Solution
The normal DLQ mechanism offered by Spring Cloud Stream will not help when Kafka consumer throws an irrecoverable deserialization exception.
This is because, this exception happens even before the consumer's `poll()` method returns.
@@ -249,16 +249,16 @@ Keep in mind that none of the dlq properties above are relevant for the discussi
They are purely meant for addressing any application level errors only.
[[basic-offset-management-in-kafka-binder]]
=== Basic offset management in Kafka binder
== Basic offset management in Kafka binder
[[problem-statement]]
==== Problem Statement
=== Problem Statement
I want to write a Spring Cloud Stream Kafka consumer application and not sure about how it manages Kafka consumer offsets.
Can you explain?
[[solution]]
==== Solution
=== Solution
We encourage you read the https://docs.spring.io/spring-cloud-stream-binder-kafka/docs/current/reference/html/spring-cloud-stream-binder-kafka.html#reset-offsets[docs] section on this to get a thorough understanding on it.
@@ -285,16 +285,16 @@ Then make sure you provide the `startOffset` value (either `earliest` or `latest
When you do that and then start the consumer application, each time you start, it starts as if this is starting for the first time and ignore any committed offsets for the partition.
[[seeking-to-arbitrary-offsets-in-kafka]]
=== Seeking to arbitrary offsets in Kafka
== Seeking to arbitrary offsets in Kafka
[[problem-statement]]
==== Problem Statement
=== Problem Statement
Using Kafka binder, I know that it can set the offset to either `earliest` or `latest`, but I have a requirement to seek the offset to something in the middle, an arbitrary offset.
Is there a way to achieve this using Spring Cloud Stream Kafka binder?
[[solution]]
==== Solution
=== Solution
Previously we saw how Kafka binder allows you to tackle basic offset management.
By default, the binder does not allow you to rewind to an arbitrary offset, at least through the mechanism we saw in that recipe.
@@ -377,7 +377,7 @@ Real world use cases are much more complex than this and you need to adjust acco
When consumer `seek` fails, it may throw some runtime exceptions and you need to decide what to do in those cases.
[[what-if-we-start-a-second-consumer-with-the-same-group-id?]]
==== What if we start a second consumer with the same group id?
=== What if we start a second consumer with the same group id?
When we add a second consumer, a rebalance will occur and some partitions will be moved around.
Let's say that the new consumer gets partitions `2` and `3`.
@@ -388,16 +388,16 @@ However, for this consumer it was simply a rebalance event and not considered as
Thus, it will not re-seek to the given offsets because of the conditional check on the `initial` argument.
[[how-do-i-manually-acknowledge-using-kafka-binder?]]
=== How do I manually acknowledge using Kafka binder?
== How do I manually acknowledge using Kafka binder?
[[problem-statement]]
==== Problem Statement
=== Problem Statement
Using Kafka binder, I want to manually acknowledge messages in my consumer.
How do I do that?
[[solution]]
==== Solution
=== Solution
By default, Kafka binder delegates to the default commit settings in Spring for Apache Kafka project.
The default `ackMode` in Spring Kafka is `batch`.
@@ -427,15 +427,15 @@ public Consumer<Message<String>> myConsumer() {
Then you set the property `spring.cloud.stream.kafka.bindings.myConsumer-in-0.consumer.ackMode` to `MANUAL` or `MANUAL_IMMEDIATE`.
[[how-do-i-override-the-default-binding-names-in-spring-cloud-stream?]]
=== How do I override the default binding names in Spring Cloud Stream?
== How do I override the default binding names in Spring Cloud Stream?
[[problem-statement]]
==== Problem Statement
=== Problem Statement
Spring Cloud Stream creates default bindings based on the function definition and signature, but how do I override these to more domain friendly names?
[[solution]]
==== Solution
=== Solution
Assume that following is your function signature.
@@ -491,15 +491,15 @@ You can easily do that by simply setting these three properties
Once you do that, it overrides the default binding names and any properties that you want to set on them must be on these new binding names.
[[how-do-i-send-a-message-key-as-part-of-my-record?]]
=== How do I send a message key as part of my record?
== How do I send a message key as part of my record?
[[problem-statement]]
==== Problem Statement
=== Problem Statement
I need to send a key along with the payload of the record, is there a way to do that in Spring Cloud Stream?
[[solution]]
==== Solution
=== Solution
It is often necessary that you want to send associative data structure like a map as the record with a key and value.
Spring Cloud Stream allows you to do that in a straightforward manner.
@@ -528,17 +528,17 @@ Please note that we use the binding name `supplier-out-0` since that is our func
Then, we use this new key when we produce the message.
[[how-do-i-use-native-serializer-and-deserializer-instead-of-message-conversion-done-by-spring-cloud-stream?]]
=== How do I use native serializer and deserializer instead of message conversion done by Spring Cloud Stream?
== How do I use native serializer and deserializer instead of message conversion done by Spring Cloud Stream?
[[problem-statement]]
==== Problem Statement
=== Problem Statement
Instead of using the message converters in Spring Cloud Stream, I want to use native Serializer and Deserializer in Kafka.
By default, Spring Cloud Stream takes care of this conversion using its internal built-in message converters.
How can I bypass this and delegate the responsibility to Kafka?
[[solution]]
==== Solution
=== Solution
This is really easy to do.
@@ -585,17 +585,17 @@ spring.cloud.stream.kafka.bindings.<binding-name>.consumer.useNativeDecoding: tr
However, in the case of Kafka binder, this is unnecessary, as by the time it reaches the binder, Kafka already deserializes them using the configured deserializers.
[[explain-how-offset-resetting-work-in-kafka-streams-binder]]
=== Explain how offset resetting work in Kafka Streams binder
== Explain how offset resetting work in Kafka Streams binder
[[problem-statement]]
==== Problem Statement
=== Problem Statement
By default, Kafka Streams binder always starts from the earliest offset for a new consumer.
Sometimes, it is beneficial or required by the application to start from the latest offset.
Kafka Streams binder allows you to do that.
[[solution]]
==== Solution
=== Solution
Before we look at the solution, let us look at the following scenario.
@@ -623,15 +623,15 @@ If you want only one binding to start from the `latest` offset and the other to
Keep in mind that, once there are committed offsets, these setting are *not* honored and the committed offsets take precedence.
[[keeping-track-of-successful-sending-of-records-producing-in-kafka]]
=== Keeping track of successful sending of records (producing) in Kafka
== Keeping track of successful sending of records (producing) in Kafka
[[problem-statement]]
==== Problem Statement
=== Problem Statement
I have a Kafka producer application and I want to keep track of all my successful sendings.
[[solution]]
==== Solution
=== Solution
Let us assume that we have this following supplier in the application.
@@ -673,15 +673,15 @@ In the `handle` method, the payload is what got sent to Kafka and the message he
Its value is a `RecordMetadata` that contains information about topic partition, current offset etc.
[[adding-custom-header-mapper-in-kafka]]
=== Adding custom header mapper in Kafka
== Adding custom header mapper in Kafka
[[problem-statement]]
==== Problem Statement
=== Problem Statement
I have a Kafka producer application that sets some headers, but they are missing in the consumer application. Why is that?
[[solution]]
==== Solution
=== Solution
Under normal circumstances, this should be fine.
@@ -751,7 +751,7 @@ public KafkaHeaderMapper kafkaBinderHeaderMapper() {
That will properly populate the `foo` header from the producer to consumer.
[[special-note-on-the-id-header]]
==== Special note on the id header
=== Special note on the id header
In Spring Cloud Stream, the `id` header is a special header, but some applications may want to have special custom id headers - something like `custom-id` or `ID` or `Id`.
The first one (`custom-id`) will propagate without any custom header mapper from producer to consumer.
@@ -792,17 +792,17 @@ public KafkaHeaderMapper kafkaBinderHeaderMapper1() {
By doing this, both `id` and `Id` headers will be available from the producer to the consumer side.
[[producing-to-multiple-topics-in-transaction]]
=== Producing to multiple topics in transaction
== Producing to multiple topics in transaction
[[problem-statement]]
==== Problem Statement
=== Problem Statement
How do I produce transactional messages to multiple Kafka topics?
For more context, see this https://stackoverflow.com/questions/68928091/dlq-bounded-retry-and-eos-when-producing-to-multiple-topics-using-spring-cloud[StackOverflow question].
[[solution]]
==== Solution
=== Solution
Use transactional support in Kafka binder for transactions and then provide an `AfterRollbackProcessor`.
In order to produce to multiple topics, use `StreamBridge` API.
@@ -844,7 +844,7 @@ DefaultAfterRollbackProcessor rollbackProcessor(KafkaTemplate<byte[], byte[]> te
```
[[required-configuration]]
==== Required Configuration
=== Required Configuration
```
spring.cloud.stream.kafka.binder.transaction.transaction-id-prefix: tx-
@@ -882,15 +882,15 @@ If you are using Spring Cloud Stream to test it by adding more consumer function
This https://stackoverflow.com/questions/68941306/spring-cloud-stream-database-transaction-does-not-roll-back[StackOverflow thread] is also related to this discussion.
[[pitfalls-to-avoid-when-running-multiple-pollable-consumers]]
=== Pitfalls to avoid when running multiple pollable consumers
== Pitfalls to avoid when running multiple pollable consumers
[[problem-statement]]
==== Problem Statement
=== Problem Statement
How can I run multiple instances of the pollable consumers and generate unique `client.id` for each instance?
[[solution]]
==== Solution
=== Solution
Assuming that I have the following definition:

10
docs/modules/ROOT/pages/preface.adoc
View File

@@ -1,5 +1,5 @@
[[a-brief-history-of-spring-s-data-integration-journey]]
=== A Brief History of Spring's Data Integration Journey
= A Brief History of Spring's Data Integration Journey
Spring's journey on Data Integration started with https://projects.spring.io/spring-integration/[Spring Integration]. With its programming model, it provided a consistent developer experience to build applications that can embrace http://www.enterpriseintegrationpatterns.com/[Enterprise Integration Patterns] to connect with external systems such as, databases, message brokers, and among others.
@@ -18,7 +18,7 @@ With Spring Cloud Stream, developers can:
- and many more. . .
[[quick-start]]
=== Quick Start
= Quick Start
You can try Spring Cloud Stream in less than 5 min even before you jump into any details by following this three-step guide.
@@ -34,7 +34,7 @@ The three steps are as follows:
. <<spring-cloud-stream-preface-adding-message-handler>>
[[spring-cloud-stream-preface-creating-sample-application]]
==== Creating a Sample Application by Using Spring Initializr
== Creating a Sample Application by Using Spring Initializr
To get started, visit the https://start.spring.io[Spring Initializr]. From there, you can generate our `LoggingConsumer` application. To do so:
. In the *Dependencies* section, start typing `stream`.
@@ -68,7 +68,7 @@ TIP: We encourage you to explore the many possibilities available in the Spring
It lets you create many different kinds of Spring applications.
[[spring-cloud-stream-preface-importing-project]]
==== Importing the Project into Your IDE
== Importing the Project into Your IDE
Now you can import the project into your IDE.
Keep in mind that, depending on the IDE, you may need to follow a specific import procedure.
@@ -81,7 +81,7 @@ It is already a valid Spring Boot application.
However, it does not do anything, so we want to add some code.
[[spring-cloud-stream-preface-adding-message-handler]]
==== Adding a Message Handler, Building, and Running
== Adding a Message Handler, Building, and Running
Modify the `com.example.loggingconsumer.LoggingConsumerApplication` class to look as follows:

10
docs/modules/ROOT/pages/rabbit/rabbit_dlq.adoc
View File

@@ -1,5 +1,5 @@
[[rabbit-dlq-processing]]
== Dead-Letter Queue Processing
= Dead-Letter Queue Processing
Because you cannot anticipate how users would want to dispose of dead-lettered messages, the framework does not provide any standard mechanism to handle them.
If the reason for the dead-lettering is transient, you may wish to route the messages back to the original queue.
@@ -13,7 +13,7 @@ You could also use `RabbitTemplate.receive()` in a batch process.
The examples assume the original destination is `so8400in` and the consumer group is `so8400`.
[[non-partitioned-destinations]]
=== Non-Partitioned Destinations
== Non-Partitioned Destinations
The first two examples are for when the destination is *not* partitioned:
@@ -126,12 +126,12 @@ public class ReRouteDlqApplication {
----
[[partitioned-destinations]]
=== Partitioned Destinations
== Partitioned Destinations
With partitioned destinations, there is one DLQ for all partitions. We determine the original queue from the headers.
[[republishtodlq=false]]
==== `republishToDlq=false`
=== `republishToDlq=false`
When `republishToDlq` is `false`, RabbitMQ publishes the message to the DLX/DLQ with an `x-death` header containing information about the original destination, as shown in the following example:
@@ -189,7 +189,7 @@ public class ReRouteDlqApplication {
----
[[republishtodlq=true]]
==== `republishToDlq=true`
=== `republishToDlq=true`
When `republishToDlq` is `true`, the republishing recoverer adds the original exchange and routing key to headers, as shown in the following example:

42
docs/modules/ROOT/pages/rabbit/rabbit_overview.adoc
View File

@@ -5,7 +5,7 @@ It contains information about its design, usage and configuration options, as we
--
[[usage]]
== Usage
= Usage
To use the RabbitMQ binder, you can add it to your Spring Cloud Stream application, by using the following Maven coordinates:
@@ -28,7 +28,7 @@ Alternatively, you can use the Spring Cloud Stream RabbitMQ Starter, as follows:
----
[[rabbitmq-binder-overview]]
== RabbitMQ Binder Overview
= RabbitMQ Binder Overview
The following simplified diagram shows how the RabbitMQ binder operates:
@@ -77,14 +77,14 @@ Starting with version 2.0, the `RabbitMessageChannelBinder` sets the `RabbitTemp
NOTE: Currently, a `multiplex` consumer (a single consumer listening to multiple queues) is only supported for message-driven consumers; polled consumers can only retrieve messages from a single queue.
[[configuration-options]]
== Configuration Options
= Configuration Options
This section contains settings specific to the RabbitMQ Binder and bound channels.
For general binding configuration options and properties, see the https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/current/reference/html/spring-cloud-stream.html#_configuration_options[Spring Cloud Stream core documentation].
[[rabbit-binder-properties]]
=== RabbitMQ Binder Properties
== RabbitMQ Binder Properties
By default, the RabbitMQ binder uses Spring Boot's `ConnectionFactory`.
Conseuqently, it supports all Spring Boot configuration options for RabbitMQ.
@@ -121,7 +121,7 @@ The name is this prefix followed by `#n`, where `n` increments each time a new c
Default: none (Spring AMQP default).
[[rabbitmq-consumer-properties]]
=== RabbitMQ Consumer Properties
== RabbitMQ Consumer Properties
The following properties are available for Rabbit consumers only and must be prefixed with `spring.cloud.stream.rabbit.bindings.<channelName>.consumer.`.
@@ -419,7 +419,7 @@ Not supported when the `containerType` is `direct`.
Default: `1`.
[[rabbitmq-stream-consumer]]
=== Initial Consumer Support for the RabbitMQ Stream Plugin
== Initial Consumer Support for the RabbitMQ Stream Plugin
Basic support for the https://rabbitmq.com/stream.html[RabbitMQ Stream Plugin] is now provided.
To enable this feature, you must add the `spring-rabbit-stream` jar to the class path - it must be the same version as `spring-amqp` and `spring-rabbit`.
@@ -473,7 +473,7 @@ public Consumer<Message<?>> input() {
Refer to the https://rabbitmq.github.io/rabbitmq-stream-java-client/stable/htmlsingle/[RabbitMQ Stream Java Client documentation] for information about configuring the environment and consumer builder.
[[rabbitmq-super-stream-consumer]]
==== Consumer Support for the RabbitMQ Super Streams
=== Consumer Support for the RabbitMQ Super Streams
See https://blog.rabbitmq.com/posts/2022/07/rabbitmq-3-11-feature-preview-super-streams[Super Streams] for information about super streams.
@@ -507,14 +507,14 @@ The framework will create a super stream named `super`, with 9 partitions.
Up to 3 instances of this application can be deployed.
[[advanced-listener-container-configuration]]
=== Advanced Listener Container Configuration
== Advanced Listener Container Configuration
To set listener container properties that are not exposed as binder or binding properties, add a single bean of type `ListenerContainerCustomizer` to the application context.
The binder and binding properties will be set and then the customizer will be called.
The customizer (`configure()` method) is provided with the queue name as well as the consumer group as arguments.
[[advanced-queue/exchange/binding-configuration]]
=== Advanced Queue/Exchange/Binding Configuration
== Advanced Queue/Exchange/Binding Configuration
From time to time, the RabbitMQ team add new features that are enabled by setting some argument when declaring, for example, a queue.
Generally, such features are enabled in the binder by adding appropriate properties, but this may not be immediately available in a current version.
@@ -522,18 +522,18 @@ Starting with version 3.0.1, you can now add `DeclarableCustomizer` bean(s) to t
This allows you to add arguments that are not currently directly supported by the binder.
[[rabbit-receiving-batch]]
=== Receiving Batched Messages
== Receiving Batched Messages
With the RabbitMQ binder, there are two types of batches handled by consumer bindings:
[[batches-created-by-producers]]
==== Batches Created by Producers
=== Batches Created by Producers
Normally, if a producer binding has `batch-enabled=true` (see <<rabbit-prod-props>>), or a message is created by a `BatchingRabbitTemplate`, elements of the batch are returned as individual calls to the listener method.
Starting with version 3.0, any such batch can be presented as a `List<?>` to the listener method if `spring.cloud.stream.bindings.<name>.consumer.batch-mode` is set to `true`.
[[consumer-side-batching]]
==== Consumer-side Batching
=== Consumer-side Batching
Starting with version 3.1, the consumer can be configured to assemble multiple inbound messages into a batch which is presented to the application as a `List<?>` of converted payloads.
The following simple application demonstrates how to use this technique:
@@ -709,7 +709,7 @@ Thing [field=value2] myHeader=headerValue2
====
[[rabbit-prod-props]]
=== Rabbit Producer Properties
== Rabbit Producer Properties
The following properties are available for Rabbit producers only and must be prefixed with `spring.cloud.stream.rabbit.bindings.<channelName>.producer.`.
@@ -1022,7 +1022,7 @@ NOTE: In the case of RabbitMQ, content type headers can be set by external appli
Spring Cloud Stream supports them as part of an extended internal protocol used for any type of transport -- including transports, such as Kafka (prior to 0.11), that do not natively support headers.
[[publisher-confirms]]
=== Publisher Confirms
== Publisher Confirms
There are two mechanisms to get the result of publishing a message; in each case, the connection factory must have `publisherConfirmType` set `ConfirmType.CORRELATED`.
The "legacy" mechanism is to set the `confirmAckChannel` to the bean name of a message channel from which you can retrieve the confirmations asynchronously; negative acks are sent to the error channel (if enabled) - see <<rabbit-error-channels>>.
@@ -1139,7 +1139,7 @@ IMPORTANT: The correlation data must be provided with a unique `id` so that the
You cannot set both `useConfirmHeader` and `confirmAckChannel` but you can still receive returned messages in the error channel when `useConfirmHeader` is true, but using the correlation header is more convenient.
[[rabbitmq-stream-producer]]
=== Initial Producer Support for the RabbitMQ Stream Plugin
== Initial Producer Support for the RabbitMQ Stream Plugin
Basic support for the https://rabbitmq.com/stream.html[RabbitMQ Stream Plugin] is now provided.
To enable this feature, you must add the `spring-rabbit-stream` jar to the class path - it must be the same version as `spring-amqp` and `spring-rabbit`.
@@ -1169,7 +1169,7 @@ ProducerMessageHandlerCustomizer<MessageHandler> handlerCustomizer() {
Refer to the https://rabbitmq.github.io/rabbitmq-stream-java-client/stable/htmlsingle/[RabbitMQ Stream Java Client documentation] for information about configuring the environment and producer builder.
[[rabbitmq-super-stream-producer]]
==== Producer Support for the RabbitMQ Super Streams
=== Producer Support for the RabbitMQ Super Streams
See https://blog.rabbitmq.com/posts/2022/07/rabbitmq-3-11-feature-preview-super-streams[Super Streams] for information about super streams.
@@ -1207,7 +1207,7 @@ spring.cloud.stream.rabbit.bindings.output.producer.declare-exchange=false
When using the stream client, if you set a `confirmAckChannel`, a copy of a successfully sent message will be sent to that channel.
[[using-existing-queues/exchanges]]
== Using Existing Queues/Exchanges
= Using Existing Queues/Exchanges
By default, the binder will automatically provision a topic exchange with the name being derived from the value of the destination binding property `<prefix><destination>`.
The destination defaults to the binding name, if not provided.
@@ -1237,7 +1237,7 @@ Refer to the property documentation above for more information.
There are similar properties used when declaring a dead-letter exchange/queue, when `autoBindDlq` is `true`.
[[retry-with-the-rabbitmq-binder]]
== Retry With the RabbitMQ Binder
= Retry With the RabbitMQ Binder
When retry is enabled within the binder, the listener container thread is suspended for any back off periods that are configured.
This might be important when strict ordering is required with a single consumer. However, for other use cases, it prevents other messages from being processed on that thread.
@@ -1261,7 +1261,7 @@ Fortunately, RabbitMQ provides the `x-death` header, which lets you determine ho
To acknowledge a message after giving up, throw an `ImmediateAcknowledgeAmqpException`.
[[putting-it-all-together]]
=== Putting it All Together
== Putting it All Together
The following configuration creates an exchange `myDestination` with queue `myDestination.consumerGroup` bound to a topic exchange with a wildcard routing key `#`:
@@ -1309,7 +1309,7 @@ public class XDeathApplication {
Notice that the count property in the `x-death` header is a `Long`.
[[rabbit-error-channels]]
== Error Channels
= Error Channels
Starting with version 1.3, the binder unconditionally sends exceptions to an error channel for each consumer destination and can also be configured to send async producer send failures to an error channel.
See "`<<spring-cloud-stream-overview-error-handling>>`" for more information.
@@ -1353,7 +1353,7 @@ There is no automatic handling of these exceptions (such as sending to a <<rabbi
You can consume these exceptions with your own Spring Integration flow.
[[rabbit-binder-health-indicator]]
== Rabbit Binder Health Indicator
= Rabbit Binder Health Indicator
The health indicator for Rabbit binder delegates to the one provided from Spring Boot.
For more information on this, see https://docs.spring.io/spring-boot/docs/current/reference/htmlsingle/#actuator.endpoints.health.auto-configured-health-indicators[this].

2
docs/modules/ROOT/pages/rabbit/rabbit_partitions.adoc
View File

@@ -1,5 +1,5 @@
[[partitioning-with-the-rabbitmq-binder]]
== Partitioning with the RabbitMQ Binder
= Partitioning with the RabbitMQ Binder
RabbitMQ does not support partitioning natively.

195
docs/modules/ROOT/pages/spring-cloud-stream.adoc
View File

@@ -11,7 +11,8 @@
// ======================================================================================
[[preface]]
== Preface
= Preface
include::preface.adoc[]
[partintro]
@@ -21,7 +22,7 @@ It covers topics such as creating and running stream applications.
--
[[spring-cloud-stream-overview-introducing]]
== Introducing Spring Cloud Stream
= Introducing Spring Cloud Stream
Spring Cloud Stream is a framework for building message-driven microservice applications.
Spring Cloud Stream builds upon Spring Boot to create standalone, production-grade Spring applications and uses Spring Integration to provide connectivity to message brokers.
@@ -72,7 +73,7 @@ class BootTestStreamApplicationTests {
----
[[main-concepts]]
== Main Concepts
= Main Concepts
Spring Cloud Stream provides a number of abstractions and primitives that simplify the writing of message-driven microservice applications.
This section gives an overview of the following:
@@ -85,7 +86,7 @@ This section gives an overview of the following:
* <<spring-cloud-stream-overview-binder-api,A pluggable Binder SPI>>
[[spring-cloud-stream-overview-application-model]]
=== Application Model
== Application Model
A Spring Cloud Stream application consists of a middleware-neutral core.
The application communicates with the outside world by establishing _bindings_ between destinations
@@ -96,13 +97,13 @@ necessary to establish bindings are handled by middleware-specific _Binder_ impl
image::SCSt-with-binder.png[width=800,scaledwidth="75%",align="center"]
[[fat-jar]]
==== Fat JAR
=== Fat JAR
Spring Cloud Stream applications can be run in stand-alone mode from your IDE for testing.
To run a Spring Cloud Stream application in production, you can create an executable (or "`fat`") JAR by using the standard Spring Boot tooling provided for Maven or Gradle. See the https://docs.spring.io/spring-boot/docs/current/reference/html/howto-build.html#howto-create-an-executable-jar-with-maven[Spring Boot Reference Guide] for more details.
[[spring-cloud-stream-overview-binder-abstraction]]
=== The Binder Abstraction
== The Binder Abstraction
Spring Cloud Stream provides Binder implementations for https://github.com/spring-cloud/spring-cloud-stream-binder-kafka[Kafka] and https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit[Rabbit MQ].
The framework also includes a test binder for integration testing of your applications as spring-cloud-stream application. See <<Testing>> section for more details.
@@ -125,7 +126,7 @@ For more complex use cases, you can also package multiple binders with your appl
[[spring-cloud-stream-overview-persistent-publish-subscribe-support]]
=== Persistent Publish-Subscribe Support
== Persistent Publish-Subscribe Support
Communication between applications follows a publish-subscribe model, where data is broadcast through shared topics.
This can be seen in the following figure, which shows a typical deployment for a set of interacting Spring Cloud Stream applications.
@@ -146,7 +147,7 @@ While the concept of publish-subscribe messaging is not new, Spring Cloud Stream
By using native middleware support, Spring Cloud Stream also simplifies use of the publish-subscribe model across different platforms.
[[consumer-groups]]
=== Consumer Groups
== Consumer Groups
While the publish-subscribe model makes it easy to connect applications through shared topics, the ability to scale up by creating multiple instances of a given application is equally important.
When doing so, different instances of an application are placed in a competing consumer relationship, where only one of the instances is expected to handle a given message.
@@ -162,7 +163,7 @@ All groups that subscribe to a given destination receive a copy of published dat
By default, when a group is not specified, Spring Cloud Stream assigns the application to an anonymous and independent single-member consumer group that is in a publish-subscribe relationship with all other consumer groups.
[[consumer-types]]
=== Consumer Types
== Consumer Types
Two types of consumer are supported:
@@ -175,7 +176,7 @@ When you wish to control the rate at which messages are processed, you might wan
// TODO This needs more description. A sentence parallel to the last sentence of the preceding paragraph would help.
[[durability]]
==== Durability
=== Durability
Consistent with the opinionated application model of Spring Cloud Stream, consumer group subscriptions are durable.
That is, a binder implementation ensures that group subscriptions are persistent and that, once at least one subscription for a group has been created, the group receives messages, even if they are sent while all applications in the group are stopped.
@@ -191,7 +192,7 @@ When scaling up a Spring Cloud Stream application, you must specify a consumer g
Doing so prevents the application's instances from receiving duplicate messages (unless that behavior is desired, which is unusual).
[[partitioning]]
=== Partitioning Support
== Partitioning Support
Spring Cloud Stream provides support for partitioning data between multiple instances of a given application.
In a partitioned scenario, the physical communication medium (such as the broker topic) is viewed as being structured into multiple partitions.
@@ -209,7 +210,7 @@ For example, in the time-windowed average calculation example, it is important t
NOTE: To set up a partitioned processing scenario, you must configure both the data-producing and the data-consuming ends.
[[programming-model]]
== Programming Model
= Programming Model
To understand the programming model, you should be familiar with the following core concepts:
@@ -220,7 +221,7 @@ To understand the programming model, you should be familiar with the following c
image::SCSt-overview.png[width=800,scaledwidth="75%",align="center"]
[[destination-binders]]
=== Destination Binders
== Destination Binders
Destination Binders are extension components of Spring Cloud Stream responsible for providing the necessary configuration and implementation to facilitate
integration with external messaging systems.
@@ -234,7 +235,7 @@ While it is out of scope of this section to discuss all of the available binder
_Binding_ as a concept, does require special attention. The next section discusses it in detail.
[[bindings]]
=== Bindings
== Bindings
As stated earlier, _Bindings_ provide a bridge between the external messaging system (e.g., queue, topic etc.) and application-provided _Producers_ and _Consumers_.
@@ -268,7 +269,7 @@ binding of to destinations exposed by the provided binder following certain nami
rules to avoid extra configuration.
[[binding-and-binding-names]]
==== Binding and Binding names
=== Binding and Binding names
Binding is an abstraction that represents a bridge between sources and targets exposed by the binder and user code,
This abstraction has a name and while we try to do our best to limit configuration required to run spring-cloud-stream applications,
@@ -282,7 +283,7 @@ NOTE: If your binding name has special characters, such as the `.` character, yo
For example `spring.cloud.stream.bindings."[my.output.binding.key]".destination`.
[[functional-binding-names]]
===== Functional binding names
==== Functional binding names
Unlike the explicit naming required by annotation-based support (legacy) used in the previous versions of spring-cloud-stream, the functional
programming model defaults to a simple convention when it comes to binding names, thus greatly simplifying application configuration.
@@ -345,7 +346,7 @@ where you are clearly correlating the input of `uppercase` function to `sample-t
For more on properties and other configuration options please see <<Configuration Options>> section.
[[explicit-binding-creation]]
===== Explicit binding creation
==== Explicit binding creation
In the previous section we explained how bindings are created implicitly driven by the names of `Function`, `Supplier` or `Consumer` beans provided by your application.
However, there are times when you may need to create binding explicitly where bindings are not tied to any function. This is typically done to
@@ -380,7 +381,7 @@ As you can see we have declared two input bindings and two output bindings while
[[producing-and-consuming-messages]]
=== Producing and Consuming Messages
== Producing and Consuming Messages
You can write a Spring Cloud Stream application by simply writing functions and exposing them as `@Bean` s.
You can also use Spring Integration annotations based configuration or
@@ -388,10 +389,10 @@ Spring Cloud Stream annotation based configuration, although starting with sprin
we recommend using functional implementations.
[[spring_cloud_function]]
==== Spring Cloud Function support
=== Spring Cloud Function support
[[overview]]
===== Overview
==== Overview
Since Spring Cloud Stream v2.1, another alternative for defining _stream handlers_ and _sources_ is to use build-in
support for https://cloud.spring.io/spring-cloud-function/[Spring Cloud Function] where they can be expressed as beans of
@@ -458,7 +459,7 @@ public static class SinkFromConsumer {
----
[[suppliers-sources]]
===== Suppliers (Sources)
==== Suppliers (Sources)
`Function` and `Consumer` are pretty straightforward when it comes to how their invocation is triggered. They are triggered based
on data (events) sent to the destination they are bound to. In other words, they are classic event-driven components.
@@ -545,7 +546,7 @@ he desired behavior you can set it to `false` at which point such supplier will
the produced Flux without splitting it.
[[supplier-threading]]
====== Supplier & threading
===== Supplier & threading
NOTE: As you have learned by now, unlike `Function` and `Consumer`, which are triggered by an event (they have input data), `Supplier` does not have
any input and thus triggered by a different mechanism - _poller_, which may have an unpredictable threading mechanism. And while the details of the
@@ -556,7 +557,7 @@ For those cases we have another mechanism via `StreamBridge`, where user has mor
in <<Sending arbitrary data to an output (e.g. Foreign event-driven sources)>> section.
[[consumer-reactive]]
===== Consumer (Reactive)
==== Consumer (Reactive)
Reactive `Consumer` is a little bit special because it has a void return type, leaving framework with no reference to subscribe to.
Most likely you will not need to write `Consumer<Flux<?>>`, and instead write it as a `Function<Flux<?>, Mono<Void>>` invoking `then`
@@ -582,7 +583,7 @@ so this limitation can only be addressed by either making your consumer reactive
[[polling-configuration-properties]]
====== Polling Configuration Properties
===== Polling Configuration Properties
The following properties are exposed by Spring Cloud Stream and are prefixed with the `spring.integration.poller.`:
@@ -615,7 +616,7 @@ Default: MILLISECONDS.
For example `--spring.integration.poller.fixed-delay=2000` sets the poller interval to poll every two seconds.
[[per-binding-polling-configuration]]
====== Per-binding polling configuration
===== Per-binding polling configuration
The previous section shows how to configure a single default poller that will be applied to all bindings. While it fits well with the model of microservices spring-cloud-stream designed for where each microservice represents a single component (e.g., Supplier) and thus default poller configuration is enough, there are edge cases where
you may have several components that require different polling configurations
@@ -624,7 +625,7 @@ For such cases please use per-binding way of configuring poller. For example, as
binding using `spring.cloud.stream.bindings.supply-out-0.producer.poller..` prefix (e.g., `spring.cloud.stream.bindings.supply-out-0.producer.poller.fixed-delay=2000`).
[[sending-arbitrary-data-to-an-output-e-g-foreign-event-driven-sources]]
===== Sending arbitrary data to an output (e.g. Foreign event-driven sources)
==== Sending arbitrary data to an output (e.g. Foreign event-driven sources)
There are cases where the actual source of data may be coming from the external (foreign) system that is not a binder. For example, the
source of the data may be a classic REST endpoint. How do we bridge such source with the functional mechanism used by spring-cloud-stream?
@@ -673,7 +674,7 @@ will go through the same routine when sending output as if it was from any Funct
of consistency as with functions. This means the output type conversion, partitioning etc are honored as if it was from the output produced by functions.
[[streambridge-and-dynamic-destinations]]
====== StreamBridge and Dynamic Destinations
===== StreamBridge and Dynamic Destinations
`StreamBridge` can also be used for cases when output destination(s) are not known ahead of time similar to the use cases
described in <<Routing FROM Consumer>> section.
@@ -750,13 +751,13 @@ By showing two examples we want to emphasize the approach will work with any typ
NOTE: If you are using the Solace PubSub+ binder, Spring Cloud Stream has reserved the `scst_targetDestination` header (retrievable via BinderHeaders.TARGET_DESTINATION), which allows for messages to be redirected from their bindings' configured destination to the target destination specified by this header. This allows for the binder to manage the resources necessary to publish to dynamic destinations, relieving the framework from having to do so, and avoids the caching issues mentioned in the previous Note. More info https://github.com/SolaceProducts/solace-spring-cloud/tree/master/solace-spring-cloud-starters/solace-spring-cloud-stream-starter#dynamic-producer-destinations[here].
[[output-content-type-with-streambridge]]
====== Output Content Type with StreamBridge
===== Output Content Type with StreamBridge
You can also provide specific content type if necessary with the following method signature `public boolean send(String bindingName, Object data, MimeType outputContentType)`.
Or if you send data as a `Message`, its content type will be honored.
[[using-specific-binder-type-with-streambridge]]
====== Using specific binder type with StreamBridge
===== Using specific binder type with StreamBridge
Spring Cloud Stream supports multiple binder scenarios. For example you may be receiving data from Kafka and sending it to RabbitMQ.
@@ -778,7 +779,7 @@ NOTE: For cases where `spring.cloud.stream.output-bindings` property is used or
argument will have no effect.
[[using-channel-interceptors-with-streambridge]]
====== Using channel interceptors with StreamBridge
===== Using channel interceptors with StreamBridge
Since `StreamBridge` uses a `MessageChannel` to establish the output binding, you can activate channel interceptors when sending data through `StreamBridge`.
It is up to the application to decide which channel interceptors to apply on `StreamBridge`.
@@ -844,7 +845,7 @@ With this approach, the application gets the ability to decide which interceptor
NOTE: `StreamBridge` provides a contract through the `StreamOperations` interface that contains all the `send` methods of `StreamBridge`. Therefore, applications may choose to autowire using `StreamOperations`. This is handy when it comes to unit testing code that uses `StreamBridge` by providing a mock or similar mechanisms for the `StreamOperations` interface.
[[reactive-functions-support]]
===== Reactive Functions support
==== Reactive Functions support
Since _Spring Cloud Function_ is build on top of https://projectreactor.io/[Project Reactor] there isn't much you need to do
@@ -883,7 +884,7 @@ Therefore, with reactive functions you must rely on the richness of the reactive
====
[[functional-composition]]
===== Functional Composition
==== Functional Composition
Using functional programming model you can also benefit from functional composition where you can dynamically compose complex handlers from a set of simple functions.
As an example let's add the following function bean to the application defined above
@@ -914,7 +915,7 @@ with the following property `spring.cloud.stream.function.bindings.toUpperCase|w
other configuration properties to refer to that binding name (e.g., `spring.cloud.stream.bindings.quotedUpperCaseInput.destination=myDestination`).
[[functional-composition-and-cross-cutting-concerns]]
====== Functional Composition and Cross-cutting Concerns
===== Functional Composition and Cross-cutting Concerns
Function composition effectively allows you to address complexity by breaking it down
to a set of simple and individually manageable/testable components that could still be
@@ -966,7 +967,7 @@ so the other function - `echo` - can benefit form it. The `echo` function stays
You can also see the usage of `spring.cloud.stream.function.bindings` property to simplify composed binding name.
[[functions-with-multiple-input-and-output-arguments]]
===== Functions with multiple input and output arguments
==== Functions with multiple input and output arguments
Starting with version 3.0 spring-cloud-stream provides support for functions that
have multiple inputs and/or multiple outputs (return values). What does this actually mean and
@@ -1085,7 +1086,7 @@ public void testSingleInputMultiOutput() {
----
[[multiple-functions-in-a-single-application]]
===== Multiple functions in a single application
==== Multiple functions in a single application
There may also be a need for grouping several message handlers in a single application. You would do so by
defining several functions.
@@ -1143,7 +1144,7 @@ public void testMultipleFunctions() {
----
[[batch-consumers]]
===== Batch Consumers
==== Batch Consumers
When using a `MessageChannelBinder` that supports batch listeners, and the feature is enabled for the consumer binding, you can set `spring.cloud.stream.bindings.<binding-name>.consumer.batch-mode` to `true` to enable the
entire batch of messages to be passed to the function in a `List`.
@@ -1157,7 +1158,7 @@ public Function<List<Person>, Person> findFirstPerson() {
----
[[batch-producers]]
===== Batch Producers
==== Batch Producers
You can also use the concept of batching on the producer side by returning a collection of Messages which effectively provides an
inverse effect where each message in the collection will be sent individually by the binder.
@@ -1182,7 +1183,7 @@ public Function<String, List<Message<String>>> batch() {
Each message in the returned list will be sent individually resulting in four messages sent to output destination.
[[spring-integration-flow-as-functions]]
===== Spring Integration flow as functions
==== Spring Integration flow as functions
When you implement a function, you may have complex requirements that fit the category
of https://www.enterpriseintegrationpatterns.com[Enterprise Integration Patterns] (EIP). These are best handled by using a
framework such as https://spring.io/projects/spring-integration[Spring Integration] (SI), which is a reference implementation of EIP.
@@ -1231,10 +1232,10 @@ into various patterns you can apply by merging the best of Spring Integration an
[[spring-cloud-streams-overview-using-polled-consumers]]
==== Using Polled Consumers
=== Using Polled Consumers
[[overview]]
===== Overview
==== Overview
When using polled consumers, you poll the `PollableMessageSource` on demand.
To define binding for polled consumer you need to provide `spring.cloud.stream.pollable-source` property.
@@ -1342,7 +1343,7 @@ boolean result = pollableSource.poll(received -> {
[[polled-errors]]
===== Handling Errors
==== Handling Errors
By default, an error channel is configured for the pollable source; if the callback throws an exception, an `ErrorMessage` is sent to the error channel (`<destination>.<group>.errors`); this error channel is also bridged to the global Spring Integration `errorChannel`.
@@ -1353,7 +1354,7 @@ If the service activator throws a `RequeueCurrentMessageException`, the message
If the listener throws a `RequeueCurrentMessageException` directly, the message will be requeued, as discussed above, and will not be sent to the error channels.
[[event-routing]]
=== Event Routing
== Event Routing
Event Routing, in the context of Spring Cloud Stream, is the ability to either
_a) route events to a particular event subscriber_ or
@@ -1361,7 +1362,7 @@ _b) route events produced by an event subscriber to a particular destination_.
Here we'll refer to it as route ‘TO’ and route ‘FROM’.
[[routing-to-consumer]]
==== Routing TO Consumer
=== Routing TO Consumer
Routing can be achieved by relying on `RoutingFunction` available in Spring Cloud Function 3.0. All you need to do is enable it via
`--spring.cloud.stream.function.routing.enabled=true` application property or provide `spring.cloud.function.routing-expression` property.
Once enabled `RoutingFunction` will be bound to input destination
@@ -1375,7 +1376,7 @@ Instruction could be provided with individual messages as well as application pr
Here are couple of samples:
[[using-message-headers]]
===== Using message headers
==== Using message headers
[source,java]
----
@SpringBootApplication
@@ -1412,7 +1413,7 @@ setting `spring.cloud.function.routing-expression` header to value `T(java.lang.
Also, for SpEL, the _root object_ of the evaluation context is `Message` so you can do evaluation on individual headers (or message) as well `....routing-expression=headers['type']`
[[using-application-properties]]
===== Using application properties
==== Using application properties
The `spring.cloud.function.routing-expression` and/or `spring.cloud.function.definition`
can be passed as application properties (e.g., `spring.cloud.function.routing-expression=headers['type']`.
@@ -1441,7 +1442,7 @@ IMPORTANT: Passing instructions via application properties is especially importa
function is only invoked once to pass the Publisher, so access to the individual items is limited.
[[routing-function-and-output-binding]]
===== Routing Function and output binding
==== Routing Function and output binding
`RoutingFunction` is a `Function` and as such treated no differently than any other function. Well. . . almost.
@@ -1458,14 +1459,14 @@ the output, the output binding for the `RoutingFunction` will be create dynamica
with regards to bindings (having both input and output bindings).
[[routing-from-consumer]]
==== Routing FROM Consumer
=== Routing FROM Consumer
Aside from static destinations, Spring Cloud Stream lets applications send messages to dynamically bound destinations.
This is useful, for example, when the target destination needs to be determined at runtime.
Applications can do so in one of two ways.
[[spring-cloud-stream-sendto-destination]]
===== spring.cloud.stream.sendto.destination
==== spring.cloud.stream.sendto.destination
You can also delegate to the framework to dynamically resolve the output destination by specifying `spring.cloud.stream.sendto.destination` header
set to the name of the destination to be resolved.
@@ -1523,7 +1524,7 @@ NOTE: If you need to support dynamic destinations with multiple binder types, us
Also, please see <<Using StreamBridge>> section to see how yet another option (StreamBridge) can be utilized for similar cases.
[[post-processing-after-sending-message]]
=== Post processing (after sending message)
== Post processing (after sending message)
Once function is invoked, its result is sent by the framework to a target destination which effectively completes function invocation cycle.
@@ -1599,7 +1600,7 @@ the post processing functionality in this case will be invoked right after invoc
which is before the function invocation cycle is complete.
[[spring-cloud-stream-overview-error-handling]]
=== Error Handling
== Error Handling
In this section we'll explain the general idea behind error handling mechanisms provided by the framework.
We'll be using Rabbit binder as an example, since individual binders define different set
@@ -1636,7 +1637,7 @@ public Function<Flux<String>, Flux<String>> uppercase() {
----
[[drop-failed-messages]]
==== Drop Failed Messages
=== Drop Failed Messages
By default, the system provides error handlers. The first error handler will simply log error message. The second error handler is binder specific error handler
which is responsible for handling error message in the context of a specific messaging system (e.g., send to DLQ). But since no additional error handling configuration was provided (in this current scenario) this handler will not do anything. So essentially after being logged, the message will be dropped.
@@ -1644,7 +1645,7 @@ which is responsible for handling error message in the context of a specific mes
While acceptable in some cases, for most cases, it is not, and we need some recovery mechanism to avoid message loss.
[[handle-error-messages]]
==== Handle Error Messages
=== Handle Error Messages
In the previous section we mentioned that by default messages that resulted in error are effectively logged and dropped. The framework also exposes mechanism for you
to provide custom error handler (i.e., to send notification or write to database, etc). You can do so by adding `Consumer` that is specifically designed to accept `ErrorMessage` which aside form all the information about the error (e.g., stack trace etc) contains the original message (the one that triggered the error).
@@ -1683,7 +1684,7 @@ If you want to have a single error handler for all function beans, you can use t
[[dlq-dead-letter-queue]]
==== DLQ - Dead Letter Queue
=== DLQ - Dead Letter Queue
Perhaps the most common mechanism, DLQ allows failed messages to be sent to a special destination: the _Dead Letter Queue_.
@@ -1749,7 +1750,7 @@ You can also facilitate immediate dispatch to DLQ (without re-tries) by setting
----
[[retry-template]]
==== Retry Template
=== Retry Template
In this section we cover configuration properties relevant to configuration of retry capabilities.
@@ -1810,13 +1811,13 @@ spring.cloud.stream.bindings.<foo>.consumer.retry-template-name=<your-retry-temp
[[spring-cloud-stream-overview-binders]]
== Binders
= Binders
Spring Cloud Stream provides a Binder abstraction for use in connecting to physical destinations at the external middleware.
This section provides information about the main concepts behind the Binder SPI, its main components, and implementation-specific details.
[[producers-and-consumers]]
=== Producers and Consumers
== Producers and Consumers
The following image shows the general relationship of producers and consumers:
@@ -1834,7 +1835,7 @@ Each group that is represented by consumer bindings for a given destination rece
If there are multiple consumer instances bound with the same group name, then messages are load-balanced across those consumer instances so that each message sent by a producer is consumed by only a single consumer instance within each group (that is, it follows normal queueing semantics).
[[spring-cloud-stream-overview-binder-api]]
=== Binder SPI
== Binder SPI
The Binder SPI consists of a number of interfaces, out-of-the box utility classes, and discovery strategies that provide a pluggable mechanism for connecting to external middleware.
@@ -1872,13 +1873,13 @@ in details, with an example, a set of steps necessary to implement a custom bind
The steps are also highlighted in the `<<spring-cloud-stream-overview-custom-binder-impl>>` section.
[[binder-detection]]
=== Binder Detection
== Binder Detection
Spring Cloud Stream relies on implementations of the Binder SPI to perform the task of connecting (binding) user code to message brokers.
Each Binder implementation typically connects to one type of messaging system.
[[classpath-detection]]
==== Classpath Detection
=== Classpath Detection
By default, Spring Cloud Stream relies on Spring Boot's auto-configuration to configure the binding process.
If a single Binder implementation is found on the classpath, Spring Cloud Stream automatically uses it.
@@ -1895,7 +1896,7 @@ For example, a Spring Cloud Stream project that aims to bind only to RabbitMQ ca
For the specific Maven coordinates of other binder dependencies, see the documentation of that binder implementation.
[[multiple-binders]]
=== Multiple Binders on the Classpath
== Multiple Binders on the Classpath
When multiple binders are present on the classpath, the application must indicate which binder is to be used for each destination binding.
Each binder configuration contains a `META-INF/spring.binders` file, which is a simple properties file, as shown in the following example:
@@ -1919,7 +1920,7 @@ spring.cloud.stream.bindings.output.binder=rabbit
----
[[multiple-systems]]
=== Connecting to Multiple Systems
== Connecting to Multiple Systems
By default, binders share the application's Spring Boot auto-configuration, so that one instance of each binder found on the classpath is created.
If your application should connect to more than one broker of the same type, you can specify multiple binder configurations, each with different environment settings.
@@ -1982,7 +1983,7 @@ environment:
----
[[binder-customizer]]
=== Customizing binders in multi binder applications
== Customizing binders in multi binder applications
When an application has multiple binders in it and wants to customize the binders, then that can be achieved by providing a `BinderCustomizer` implementation.
In the case of applications with a single binder, this special customizer is not necessary since the binder context can access the customization beans directly.
@@ -2014,11 +2015,11 @@ public BinderCustomizer binderCustomizer() {
Note that, when there are more than one instance of the same type of the binder, the binder name can be used to filter customization.
[[binding_visualization_control]]
=== Binding visualization and control
== Binding visualization and control
Spring Cloud Stream supports visualization and control of the Bindings through Actuator endpoints as well as programmatic way.
[[programmatic-way]]
==== Programmatic way
=== Programmatic way
Since version 3.1 we expose `org.springframework.cloud.stream.binding.BindingsLifecycleController` which is registered as bean and once
injected could be used to control the lifecycle of individual bindings
@@ -2038,7 +2039,7 @@ assertThat(binding.isRunning()).isFalse();
----
[[actuator]]
==== Actuator
=== Actuator
Since actuator and web are optional, you must first add one of the web dependencies as well as add the actuator dependency manually.
The following example shows how to add the dependency for the Web framework:
@@ -2097,7 +2098,7 @@ NOTE: `PAUSED` and `RESUMED` work only when the corresponding binder and its und
Currently, only Kafka and [Solace](https://github.com/SolaceProducts/solace-spring-cloud/tree/master/solace-spring-cloud-starters/solace-spring-cloud-stream-starter#consumer-bindings-pauseresume) binders supports the `PAUSED` and `RESUMED` states.
[[binder-configuration-properties]]
=== Binder Configuration Properties
== Binder Configuration Properties
The following properties are available when customizing binder configurations. These properties exposed via `org.springframework.cloud.stream.config.BinderProperties`
@@ -2125,7 +2126,7 @@ This setting allows adding binder configurations without interfering with the de
Default: `true`.
[[spring-cloud-stream-overview-custom-binder-impl]]
=== Implementing Custom Binders
== Implementing Custom Binders
In order to implement a custom `Binder`, all you need is to:
@@ -2372,7 +2373,7 @@ com.example.springcloudstreamcustombinder.config.FileMessageBinderConfiguration
----
[[configuration-options]]
== Configuration Options
= Configuration Options
Spring Cloud Stream supports general configuration options as well as configuration for bindings and binders.
Some binders let additional binding properties support middleware-specific features.
@@ -2381,7 +2382,7 @@ Configuration options can be provided to Spring Cloud Stream applications throug
This includes application arguments, environment variables, and YAML or .properties files.
[[binding-service-properties]]
=== Binding Service Properties
== Binding Service Properties
These properties are exposed via `org.springframework.cloud.stream.config.BindingServiceProperties`
@@ -2423,7 +2424,7 @@ Set it to zero to treat such conditions as fatal, preventing the application fro
Default: `30`
[[binding-properties]]
=== Binding Properties
== Binding Properties
Binding properties are supplied by using the format of `spring.cloud.stream.bindings.<bindingName>.<property>=<value>`.
The `<bindingName>` represents the name of the binding being configured.
@@ -2444,7 +2445,7 @@ and `spring.cloud.stream.default.<producer|consumer>.<property>=<value>` for com
When it comes to avoiding repetitions for extended binding properties, this format should be used - `spring.cloud.stream.<binder-type>.default.<producer|consumer>.<property>=<value>`.
[[common-binding-properties]]
==== Common Binding Properties
=== Common Binding Properties
These properties are exposed via `org.springframework.cloud.stream.config.BindingProperties`
@@ -2476,7 +2477,7 @@ See `<<multiple-binders>>` for details.
Default: `null` (the default binder is used, if it exists).
[[consumer-properties]]
==== Consumer Properties
=== Consumer Properties
These properties are exposed via `org.springframework.cloud.stream.binder.ConsumerProperties`
@@ -2563,7 +2564,7 @@ When set to true, the underlying binder will natively multiplex destinations on
Default: `false`.
[[advanced-consumer-configuration]]
==== Advanced Consumer Configuration
=== Advanced Consumer Configuration
For advanced configuration of the underlying message listener container for message-driven consumers, add a single `ListenerContainerCustomizer` bean to the application context.
It will be invoked after the above properties have been applied and can be used to set additional properties.
@@ -2587,7 +2588,7 @@ public MessageSourceCustomizer<AmqpMessageSource> sourceCustomizer() {
====
[[producer-properties]]
==== Producer Properties
=== Producer Properties
These properties are exposed via `org.springframework.cloud.stream.binder.ProducerProperties`
@@ -2658,7 +2659,7 @@ Default: false.
[[advanced-producer-configuration]]
==== Advanced Producer Configuration
=== Advanced Producer Configuration
In some cases Producer Properties are not enough to properly configure a producing MessageHandler in the binder, or may be you prefer a programmatic approach
while configuring such producing MessageHandler. Regardless of the reason, spring-cloud-stream provides `ProducerMessageHandlerCustomizer` to accomplish it.
@@ -2687,7 +2688,7 @@ All you need to do is provide implementation of this strategy and configure it a
====
[[content-type-management]]
== Content Type Negotiation
= Content Type Negotiation
Data transformation is one of the core features of any message-driven microservice architecture. Given that, in Spring Cloud Stream, such data
is represented as a Spring `Message`, a message may have to be transformed to a desired shape or size before reaching its destination. This is required for two reasons:
@@ -2703,7 +2704,7 @@ In Spring Cloud Stream, message transformation is accomplished with an `org.spri
NOTE: As a supplement to the details to follow, you may also want to read the following https://spring.io/blog/2018/02/26/spring-cloud-stream-2-0-content-type-negotiation-and-transformation[blog post].
[[mechanics]]
=== Mechanics
== Mechanics
To better understand the mechanics and the necessity behind content-type negotiation, we take a look at a very simple use case by using the following message handler as an example:
@@ -2749,7 +2750,7 @@ You can always opt out of returning a `Message` from the handler method where yo
If there is an internal pipeline, the `Message` is sent to the next handler by going through the same process of conversion. However, if there is no internal pipeline or you have reached the end of it, the `Message` is sent back to the output destination.
[[content-type-versus-argument-type]]
==== Content Type versus Argument Type
=== Content Type versus Argument Type
As mentioned earlier, for the framework to select the appropriate `MessageConverter`, it requires argument type and, optionally, content type information.
The logic for selecting the appropriate `MessageConverter` resides with the argument resolvers (`HandlerMethodArgumentResolvers`), which trigger right before the invocation of the user-defined handler method (which is when the actual argument type is known to the framework).
@@ -2771,7 +2772,7 @@ Remember that the `contentType` is complementary to the target type.
If you wish, you can provide a hint, which `MessageConverter` may or may not take into consideration.
[[message-converters]]
==== Message Converters
=== Message Converters
`MessageConverters` define two methods:
@@ -2802,7 +2803,7 @@ Message<byte[]> toMessage(Object payload, @Nullable MessageHeaders headers);
----
[[provided-messageconverters]]
=== Provided MessageConverters
== Provided MessageConverters
As mentioned earlier, the framework already provides a stack of `MessageConverters` to handle most common use cases.
The following list describes the provided `MessageConverters`, in order of precedence (the first `MessageConverter` that works is used):
@@ -2817,7 +2818,7 @@ However, most likely, you found some uncommon case (such as a custom `contentTyp
does not know how to convert. If that is the case, you can add custom `MessageConverter`. See <<spring-cloud-stream-overview-user-defined-message-converters>>.
[[spring-cloud-stream-overview-user-defined-message-converters]]
=== User-defined Message Converters
== User-defined Message Converters
Spring Cloud Stream exposes a mechanism to define and register additional `MessageConverter` s.
To use it, implement `org.springframework.messaging.converter.MessageConverter`, configure it as a `@Bean`.
@@ -2861,7 +2862,7 @@ public class MyCustomMessageConverter extends AbstractMessageConverter {
----
[[inter-application-communication]]
== Inter-Application Communication
= Inter-Application Communication
Spring Cloud Stream enables communication between applications. Inter-application communication is a complex issue spanning several concerns, as described in the following topics:
@@ -2870,7 +2871,7 @@ Spring Cloud Stream enables communication between applications. Inter-applicatio
* `<<spring-cloud-stream-overview-partitioning>>`
[[spring-cloud-stream-overview-connecting-multiple-application-instances]]
=== Connecting Multiple Application Instances
== Connecting Multiple Application Instances
While Spring Cloud Stream makes it easy for individual Spring Boot applications to connect to messaging systems, the typical scenario for Spring Cloud Stream is the creation of multi-application pipelines, where microservice applications send data to each other.
You can achieve this scenario by correlating the input and output destinations of "`adjacent`" applications.
@@ -2890,7 +2891,7 @@ spring.cloud.stream.bindings.input.destination=ticktock
----
[[spring-cloud-stream-overview-instance-index-instance-count]]
=== Instance Index and Instance Count
== Instance Index and Instance Count
When scaling up Spring Cloud Stream applications, each instance can receive information about how many other instances of the same application exist and what its own instance index is.
Spring Cloud Stream does this through the `spring.cloud.stream.instanceCount` and `spring.cloud.stream.instanceIndex` properties.
@@ -2902,7 +2903,7 @@ By default, `spring.cloud.stream.instanceCount` is `1`, and `spring.cloud.stream
In a scaled-up scenario, correct configuration of these two properties is important for addressing partitioning behavior (see below) in general, and the two properties are always required by certain binders (for example, the Kafka binder) in order to ensure that data are split correctly across multiple consumer instances.
[[spring-cloud-stream-overview-partitioning]]
=== Partitioning
== Partitioning
Partitioning in Spring Cloud Stream consists of two tasks:
@@ -2910,7 +2911,7 @@ Partitioning in Spring Cloud Stream consists of two tasks:
* `<<spring-cloud-stream-overview-configuring-input-bindings-partitioning>>`
[[spring-cloud-stream-overview-configuring-output-bindings-partitioning]]
==== Configuring Output Bindings for Partitioning
=== Configuring Output Bindings for Partitioning
You can configure an output binding to send partitioned data by setting one and only one of its `partitionKeyExpression` or `partitionKeyExtractorName` properties, as well as its `partitionCount` property.
@@ -2962,7 +2963,7 @@ NOTE: In previous versions of Spring Cloud Stream you could specify the implemen
Since version 3.0, this property is removed.
[[spring-cloud-stream-overview-configuring-input-bindings-partitioning]]
==== Configuring Input Bindings for Partitioning
=== Configuring Input Bindings for Partitioning
An input binding (with the binding name `uppercase-in-0`) is configured to receive partitioned data by setting its `partitioned`
property, as well as the `instanceIndex` and `instanceCount` properties on the application itself, as shown in the following example:
@@ -2987,20 +2988,20 @@ When a binder configuration requires them, it is important to set both values co
While a scenario in which using multiple instances for partitioned data processing may be complex to set up in a standalone case, Spring Cloud Dataflow can simplify the process significantly by populating both the input and output values correctly and by letting you rely on the runtime infrastructure to provide information about the instance index and instance count.
[[testing]]
== Testing
= Testing
Spring Cloud Stream provides support for testing your microservice applications without connecting to a messaging system.
[[spring_integration_test_binder]]
=== Spring Integration Test Binder
== Spring Integration Test Binder
Spring Cloud Stream comes with a test binder which you can use for testing the various application components without requiring an actual real-world binder implementation or a message broker.
This test binder acts as a bridge between _unit_ and _integration_ testing and is based on https://spring.io/projects/spring-integration[Spring Integration] framework as an in-JVM message broker essentially giving you the best of both worlds - a real binder without the networking.
[[test-binder-configuration]]
==== Test Binder configuration
=== Test Binder configuration
To enable Spring Integration test binder, all you need is to add it as a dependency.
***Add required dependencies***
@@ -3024,7 +3025,7 @@ testImplementation("org.springframework.cloud:spring-cloud-stream-test-binder")
----
[[test-binder-usage]]
==== Test Binder usage
=== Test Binder usage
Now you can test your microservice as a simple unit test
@@ -3167,7 +3168,7 @@ public void testMultipleFunctions() {
[[test-binder-and-pollablemessagesource]]
==== Test Binder and PollableMessageSource
=== Test Binder and PollableMessageSource
Spring Integration Test Binder also allows you to write tests when working with `PollableMessageSource` (see <<Using Polled Consumers>> for more details).
The important thing that needs to be understood though is that polling is not event-driven, and that `PollableMessageSource` is a strategy which exposes operation to produce (poll for) a Message (singular).
@@ -3250,7 +3251,7 @@ NOTE: DO NOT name this bean `messageSource` as it is going to be in conflict wit
provided by Spring Boot for unrelated reasons.
[[special-note-on-mixing-test-binder-and-regular-middleware-binder-for-testing]]
==== Special Note on Mixing Test Binder and Regular Middleware Binder for Testing
=== Special Note on Mixing Test Binder and Regular Middleware Binder for Testing
The Spring Integration based test binder is provided for testing the application without involving an actual middleware based binder such as the Kafka or RabbitMQ binder.
As described in the sections above, the test binder helps you to verify the application behavior quickly by relying on the in-memory Spring Integration channels.
@@ -3259,7 +3260,7 @@ In other words, you cannot mix both the test binder and a regular middleware bin
After testing the application with the test binder, if you want to continue doing further integration tests using the actual middleware binder, it is recommended to add those tests that use the actual binder in a separate module so that those tests can make the proper connection to the actual middleware rather than relying on the in-memory channels provided by the test binder.
[[health-indicator]]
== Health Indicator
= Health Indicator
Spring Cloud Stream provides a health indicator for binders.
It is registered under the name `binders` and can be enabled or disabled by setting the `management.health.binders.enabled` property.
@@ -3293,19 +3294,19 @@ The top level application health check status will be reported as `DOWN`.
In this situation, you can simply remove the dependency for kafka binder from your application since you are not using it.
[[samples]]
== Samples
= Samples
For Spring Cloud Stream samples, see the https://github.com/spring-cloud/spring-cloud-stream-samples[spring-cloud-stream-samples] repository on GitHub.
[[deploying-stream-applications-on-cloudfoundry]]
=== Deploying Stream Applications on CloudFoundry
== Deploying Stream Applications on CloudFoundry
On CloudFoundry, services are usually exposed through a special environment variable called https://docs.cloudfoundry.org/devguide/deploy-apps/environment-variable.html#VCAP-SERVICES[VCAP_SERVICES].
When configuring your binder connections, you can use the values from an environment variable as explained on the http://docs.spring.io/spring-cloud-dataflow-server-cloudfoundry/docs/current-SNAPSHOT/reference/htmlsingle/#getting-started-ups[dataflow Cloud Foundry Server] docs.
[[binder-implementations]]
== Binder Implementations
= Binder Implementations
The following is the list of available binder implementations