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

Insert explicit ids for headers

Browse Source
This commit is contained in:
Marcin Grzejszczak
2023-09-08 16:12:37 +02:00
committed by Oleg Zhurakousky
parent 4c404c6d51
commit 53dc8e151e
18 changed files with 189 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

@@ -19,4 +19,4 @@
|spring.cloud.stream.sendto.destination | `+++none+++` | The name of the header used to determine the name of the output destination
|spring.cloud.stream.source | | A semi-colon delimited string representing the names of the sources based on which source bindings will be created. This is primarily to support cases where source binding may be required without providing a corresponding Supplier. (e.g., for cases where the actual source of data is outside of scope of spring-cloud-stream - HTTP -> Stream) @deprecated use {@link #outputBindings}
|===
|===

1
docs/modules/ROOT/pages/index.adoc
View File

@@ -1,3 +1,4 @@
[[spring-cloud-stream-reference-documentation]]
= Spring Cloud Stream Reference Documentation
Sabby Anandan; Marius Bogoevici; Eric Bottard; Mark Fisher; Ilayaperumal Gopinathan; Mark Heckler; Gunnar Hillert; Mark Pollack; Patrick Peralta; Glenn Renfro; Thomas Risberg; Dave Syer; David Turanski; Janne Valkealahti; Benjamin Klein; Vinicius Carvalho; Gary Russell; Oleg Zhurakousky; Jay Bryant; Soby Chacko; Domenico Sibilio

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

@@ -7,8 +7,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
[[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:
@@ -31,8 +33,10 @@ Alternatively, you can also use the Spring Cloud Stream Kafka Starter, as shown
</dependency>
----
[[apache-kafka-streams-binder]]
== Apache Kafka Streams Binder
[[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:
@@ -45,7 +49,9 @@ To use Apache Kafka Streams binder, you need to add `spring-cloud-stream-binder-
</dependency>
----
[[rabbitmq-binder]]
== RabbitMQ Binder
[[usage]]
=== Usage
To use the RabbitMQ binder, you can add it to your Spring Cloud Stream application, by using the following Maven coordinates:

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

@@ -1,9 +1,11 @@
[[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
Following are the maven coordinates for the reactive Kafka binder.
@@ -15,6 +17,7 @@ Following are the maven coordinates for the reactive Kafka binder.
</dependency>
```
[[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.
@@ -92,6 +95,7 @@ Note that `reactor-kafka` returns a `Flux<Flux<ConsumerRecord<?, ?>>>` when usin
Given that Spring has no access to the contents of the inner flux, the application must deal with the native `ConsumerRecord`; there is no message conversion or conversion service applied to the contents.
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
In the above `upppercase` function, we are consuming the record as `Flux<String>` and then produce it as `Flux<String>`.
@@ -158,6 +162,7 @@ Then that message payload which is the `ReceiverRecord` is provided to the user
If `reactiveAutoCommit` is `false` (default), call `rec.receiverOffset().acknowledge()` (or `commit()`) to cause the offset to be committed; if `reactiveAutoCommit` is `true`, the flux supplies `ConsumerRecord` s instead.
Refer to the `reactor-kafka` documentation and javadocs for more information.
[[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.
@@ -173,16 +178,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
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
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
Starting with version 4.0.3, you can configure the `resultMetadataChannel` to receive `SenderResult<?>` s to determine success/failure of sends.

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

@@ -5,8 +5,10 @@ It contains information about its design, usage, and configuration options, as w
In addition, this guide explains the Kafka Streams binding capabilities of Spring Cloud Stream.
--
[[apache-kafka-binder]]
== Apache Kafka Binder
[[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:
@@ -29,6 +31,7 @@ Alternatively, you can also use the Spring Cloud Stream Kafka Starter, as shown
</dependency>
----
[[overview]]
=== Overview
The following image shows a simplified diagram of how the Apache Kafka binder operates:
@@ -45,6 +48,7 @@ This client can communicate with older brokers (see the Kafka documentation), bu
For example, with versions earlier than 0.11.x.x, native headers are not supported.
Also, 0.11.x.x does not support the `autoAddPartitions` property.
[[configuration-options]]
=== Configuration Options
This section contains the configuration options used by the Apache Kafka binder.
@@ -379,6 +383,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
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.
@@ -517,10 +522,12 @@ If set to true, records published to this output binding will not be run in a tr
+
Default: `false`
[[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
This example illustrates how one may manually acknowledge offsets in a consumer application.
@@ -549,6 +556,7 @@ public class ManuallyAcknowdledgingConsumer {
}
----
[[example:-security-configuration]]
===== Example: Security Configuration
Apache Kafka 0.9 supports secure connections between client and brokers.
@@ -568,6 +576,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
The JAAS and (optionally) krb5 file locations can be set for Spring Cloud Stream applications by using system properties.
@@ -581,6 +590,7 @@ The following example shows how to launch a Spring Cloud Stream application with
--spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_PLAINTEXT
----
[[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.
@@ -637,6 +647,7 @@ Usually, applications may use principals that do not have administrative rights
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
When connecting to multiple clusters in which each one requires separate JAAS configuration, then set the JAAS configuration using the property `sasl.jaas.config`.

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

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

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

@@ -1,13 +1,17 @@
[[tips-tricks-and-recipes]]
== Tips, Tricks and Recipes
[[simple-dlq-with-kafka]]
=== Simple DLQ with Kafka
[[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
The solution for this problem is to use the DLQ feature in Spring Cloud Stream.
@@ -80,12 +84,15 @@ spring.cloud.stream:
```
[[dlq-with-advanced-retry-options]]
=== DLQ with Advanced Retry Options
[[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
If you followed the above recipe, then you get the default retry options built into the Kafka binder when the processing encounters an error.
@@ -124,14 +131,17 @@ If you have multiple `@StreamRetryTemplate` beans, then you can specify which on
spring.cloud.stream.bindings.processData-in-0.consumer.retry-template-name=<your-retry-template-bean-name>
```
[[handling-deserialization-errors-with-dlq]]
=== Handling Deserialization errors with DLQ
[[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
The normal DLQ mechanism offered by Spring Cloud Stream will not help when Kafka consumer throws an irrecoverable deserialization exception.
@@ -238,13 +248,16 @@ We are also indicating that the actual deserializer to delegate is the `StringDe
Keep in mind that none of the dlq properties above are relevant for the discussions in this recipe.
They are purely meant for addressing any application level errors only.
[[basic-offset-management-in-kafka-binder]]
=== Basic offset management in Kafka binder
[[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
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.
@@ -271,13 +284,16 @@ In order to do that, set the property `spring.cloud.stream.kafka.bindings.<bindi
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
[[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
Previously we saw how Kafka binder allows you to tackle basic offset management.
@@ -360,6 +376,7 @@ This is just a rudimentary implementation.
Real world use cases are much more complex than this and you need to adjust accordingly, but this certainly gives you a basic sketch.
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?
When we add a second consumer, a rebalance will occur and some partitions will be moved around.
@@ -370,13 +387,16 @@ In the case of the first consumer, it now only has partitions `0` and `1`
However, for this consumer it was simply a rebalance event and not considered as an intial assignment.
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?
[[problem-statement]]
==== Problem Statement
Using Kafka binder, I want to manually acknowledge messages in my consumer.
How do I do that?
[[solution]]
==== Solution
By default, Kafka binder delegates to the default commit settings in Spring for Apache Kafka project.
@@ -406,12 +426,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?
[[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
Assume that following is your function signature.
@@ -467,12 +490,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?
[[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
It is often necessary that you want to send associative data structure like a map as the record with a key and value.
@@ -501,14 +527,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?
[[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
This is really easy to do.
@@ -555,14 +584,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
[[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
Before we look at the solution, let us look at the following scenario.
@@ -590,12 +622,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
[[problem-statement]]
==== Problem Statement
I have a Kafka producer application and I want to keep track of all my successful sendings.
[[solution]]
==== Solution
Let us assume that we have this following supplier in the application.
@@ -637,12 +672,15 @@ public IntegrationFlow integrationFlow() {
In the `handle` method, the payload is what got sent to Kafka and the message headers contain a special key called `kafka_recordMetadata`.
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
[[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
Under normal circumstances, this should be fine.
@@ -712,6 +750,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
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`.
@@ -752,14 +791,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
[[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
Use transactional support in Kafka binder for transactions and then provide an `AfterRollbackProcessor`.
@@ -801,6 +843,7 @@ DefaultAfterRollbackProcessor rollbackProcessor(KafkaTemplate<byte[], byte[]> te
```
[[required-configuration]]
==== Required Configuration
```
@@ -838,12 +881,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
[[problem-statement]]
==== Problem Statement
How can I run multiple instances of the pollable consumers and generate unique `client.id` for each instance?
[[solution]]
==== Solution
Assuming that I have the following definition:

1
docs/modules/ROOT/pages/kafka/spring-cloud-stream-binder-kafka.adoc
View File

@@ -35,6 +35,7 @@ Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinat
*{project-version}*
[[reference-guide]]
= Reference Guide
include::kafka/kafka_overview.adoc[]

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

@@ -1,3 +1,4 @@
[[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.
@@ -16,6 +17,7 @@ With Spring Cloud Stream, developers can:
- Rely on the framework's automatic content-type support for common use-cases. Extending to different data conversion types is possible.
- and many more. . .
[[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.
@@ -157,6 +159,7 @@ Now you have a working (albeit very basic) Spring Cloud Stream application.
[[spel-and-streaming-data]]
[[spring-expression-language-spel-in-the-context-of-streaming-data]]
== Spring Expression Language (SpEL) in the context of Streaming data
Throughout this reference manual you will encounter many features and examples where you can utilize Spring Expression Language (SpEL). It is important to understand certain limitations when it comes to using it.

9
docs/modules/ROOT/pages/pulsar/pulsar_binder.adoc
View File

@@ -4,6 +4,7 @@
Spring for Apache Pulsar provides a binder for Spring Cloud Stream that we can use to build event-driven microservices using pub-sub paradigms.
In this section, we will go through the basic details of this binder.
[[usage]]
== Usage
We need to include the following dependency on your application to use Apache Pulsar binder for Spring Cloud Stream.
@@ -28,6 +29,7 @@ dependencies {
}
----
[[overview]]
== Overview
The Spring Cloud Stream binder for Apache Pulsar allows the applications to focus on business logic rather than dealing with the lower-level details of managing and maintaining Pulsar.
@@ -123,6 +125,7 @@ If a destination is not provided, the binding name becomes the value for the des
When running the above app, you should see that the supplier executes every second, which is then consumed by the function and enhances the time consumed by the logger consumer.
[[message-conversion-in-binder-based-applications]]
== Message Conversion in Binder-based Applications
In the above sample application, we provided no schema information for message conversion.
@@ -131,6 +134,7 @@ Unless specified, Spring Cloud Stream uses `application/json` as the `content-ty
On the outbound, the data is serialized as `byte[],` and the Pulsar binder then uses `Schema.BYTES` to send it over the wire to the Pulsar topic.
Similarly, on the inbound, the data is consumed as `byte[]` from the Pulsar topic and then converted into the target type using the proper message converter.
[[using-native-conversion-in-pulsar-using-pulsar-schema]]
=== Using Native Conversion in Pulsar using Pulsar Schema
Although the default is to use the framework-provided message conversion, Spring Cloud Stream allows each binder to determine how the message should be converted.
@@ -190,12 +194,14 @@ If you have both keys and values on the message, you can use `message-key-type`
TIP: Any configured custom schema mappings will be consulted when the `schema-type` property is omitted.
[[message-header-conversion]]
=== Message Header Conversion
Each message typically has header information that needs to be carried along as the message traverses between Pulsar and Spring Messaging via Spring Cloud Stream input and output bindings.
To support this traversal, the framework handles the necessary message header conversion.
https://docs.spring.io/spring-pulsar/docs/current-SNAPSHOT/reference/html/#_pulsar_headers[Pulsar Headers]
[[custom-header-mapper]]
==== Custom Header Mapper
The Pulsar binder is configured with a default header mapper that can be overridden by providing your own `PulsarHeaderMapper` bean.
@@ -219,6 +225,7 @@ public PulsarHeaderMapper customPulsarHeaderMapper() {
}
----
[[using-pulsar-properties-in-the-binder]]
== Using Pulsar Properties in the Binder
The binder uses basic components from Spring for Apache Pulsar framework to build its producer and consumer bindings.
@@ -240,6 +247,7 @@ This way, you can have a separate set of Pulsar properties applied for different
The highest precedence is for extended binding properties.
The precedence order of applying the properties in the binder is `extended binding properties -> binder properties -> Spring Boot properties.` (going from highest to lowest).
[[pulsar-topic-provisioner]]
== Pulsar Topic Provisioner
Spring Cloud Stream binder for Apache Pulsar comes with an out-of-the-box provisioner for Pulsar topics.
@@ -248,6 +256,7 @@ However, this is a basic non-partitioned topic, and if you want advanced feature
Pulsar topic provisioner uses `PulsarAdministration` from the framework, which uses the `PulsarAdminBuilder.`
For this reason, you need to set the `spring.pulsar.administration.service-url` property unless you are running Pulsar on the default server and port.
[[specifying-partition-count-when-creating-the-topic]]
=== Specifying partition count when creating the topic
When creating the topic, you can set the partition count in two ways.

1
docs/modules/ROOT/pages/pulsar/spring-cloud-stream-binder-pulsar.adoc
View File

@@ -19,6 +19,7 @@ Soby Chacko; Chris Bono; Alexander Preuß; Jay Bryant; Christophe Bornet
*{project-version}*
[[reference-guide]]
= Reference Guide
include::pulsar/pulsar_binder.adoc[]

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

@@ -12,6 +12,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
The first two examples are for when the destination is *not* partitioned:
@@ -124,10 +125,12 @@ public class ReRouteDlqApplication {
}
----
[[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`
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:
@@ -185,6 +188,7 @@ public class ReRouteDlqApplication {
}
----
[[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:

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

@@ -4,6 +4,7 @@ This guide describes the RabbitMQ implementation of the Spring Cloud Stream Bind
It contains information about its design, usage and configuration options, as well as information on how the Stream Cloud Stream concepts map into RabbitMQ specific constructs.
--
[[usage]]
== Usage
To use the RabbitMQ binder, you can add it to your Spring Cloud Stream application, by using the following Maven coordinates:
@@ -26,6 +27,7 @@ Alternatively, you can use the Spring Cloud Stream RabbitMQ Starter, as follows:
</dependency>
----
[[rabbitmq-binder-overview]]
== RabbitMQ Binder Overview
The following simplified diagram shows how the RabbitMQ binder operates:
@@ -74,6 +76,7 @@ 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
This section contains settings specific to the RabbitMQ Binder and bound channels.
@@ -117,6 +120,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
The following properties are available for Rabbit consumers only and must be prefixed with `spring.cloud.stream.rabbit.bindings.<channelName>.consumer.`.
@@ -502,12 +506,14 @@ spring.cloud.stream.rabbit.bindings.input-in-0.consumer.super-stream=true
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
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
From time to time, the RabbitMQ team add new features that are enabled by setting some argument when declaring, for example, a queue.
@@ -520,11 +526,13 @@ This allows you to add arguments that are not currently directly supported by th
With the RabbitMQ binder, there are two types of batches handled by consumer bindings:
[[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
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.
@@ -1198,6 +1206,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
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>`.
@@ -1227,6 +1236,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
When retry is enabled within the binder, the listener container thread is suspended for any back off periods that are configured.
@@ -1250,6 +1260,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
The following configuration creates an exchange `myDestination` with queue `myDestination.consumerGroup` bound to a topic exchange with a wildcard routing key `#`:

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

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

1
docs/modules/ROOT/pages/rabbit/spring-cloud-stream-binder-rabbit.adoc
View File

@@ -34,6 +34,7 @@ Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinat
*{project-version}*
[[reference-guide]]
= Reference Guide

12
docs/modules/ROOT/pages/schema-registry/spring-cloud-stream-schema-registry.adoc
View File

@@ -1,6 +1,7 @@
[[spring-cloud-stream-schema-registry-reference]]
= Spring Cloud Stream Schema Registry
[[introduction]]
== Introduction
When organizations have a messaging based pub/sub architecture and multiple producer and consumer microservices communicate with each other, it is often necessary for all those microservices to agree on a contract that is based on a schema.
@@ -9,6 +10,7 @@ Spring Cloud Stream provides support for a standalone schema registry server usi
Spring Cloud Stream schema registry support also provides support for avro based schema registry clients, which essentially provide message converters that communicates with the schema registry for reconciling schema during message conversion.
The schema evolution support provided by Spring Cloud Stream works both with the aforementioned standalone schema registry as well as the schema registry provided by Confluent that specifically works with Apache Kafka.
[[spring-cloud-stream-schema-registry-overview]]
==== Spring Cloud Stream Schema Registry overview
Spring Cloud Stream Schema Registry provides support for schema evolution so that the data can be evolved over time and still work with older or newer producers and consumers and vice versa.
@@ -35,6 +37,7 @@ Spring Cloud Stream Schema Registry provides the following components
Currently, the client can communicate to the standalone schema registry or the Confluent Schema Registry.
[[schema-registry-client]]
=== Schema Registry Client
The client-side abstraction for interacting with schema registry servers is the `SchemaRegistryClient` interface, which has the following structure:
@@ -70,6 +73,7 @@ Because of this, it uses a `DefaultSchemaRegistryClient` that does not cache res
If you intend to change the default behavior, you can use the client directly on your code and override it to the desired outcome.
To do so, you have to add the property `spring.cloud.stream.schemaRegistryClient.cached=true` to your application properties.
[[schema-registry-client-properties]]
==== Schema Registry Client Properties
The Schema Registry Client supports the following properties:
@@ -84,6 +88,7 @@ Clients using the schema registry client should set this to `true`.
+
Default:: `false`
[[avro-schema-registry-client-message-converters]]
=== Avro Schema Registry Client Message Converters
For applications that have a SchemaRegistryClient bean registered with the application context, Spring Cloud Stream auto-configures an Apache Avro message converter for schema management.
@@ -105,6 +110,7 @@ For example, a message of the type `User` might be sent as a binary payload with
When receiving messages, the converter infers the schema reference from the header of the incoming message and tries to retrieve it. The schema is used as the writer schema in the deserialization process.
[[avro-schema-registry-message-converter-properties]]
==== Avro Schema Registry Message Converter Properties
If you have enabled Avro based schema registry client by setting `spring.cloud.stream.stream.bindings.<output-binding-name>.contentType=application/*+avro`, you can customize the behavior of the registration by setting the following properties.
@@ -132,6 +138,7 @@ spring.cloud.stream.schema.avro.ignoreSchemaRegistryServer:: Ignore any schema r
+
Default: `false`
[[apache-avro-message-converters]]
=== Apache Avro Message Converters
Spring Cloud Stream provides support for schema-based message converters through its `spring-cloud-stream-schema-registry-client` module.
@@ -142,6 +149,7 @@ The `spring-cloud-stream-schema-registry-client` module contains two types of me
* Converters that use the class information of the serialized or deserialized objects or a schema with a location known at startup.
* Converters that use a schema registry. They locate the schemas at runtime and dynamically register new schemas as domain objects evolve.
[[converters-with-schema-support]]
=== Converters with Schema Support
The `AvroSchemaMessageConverter` supports serializing and deserializing messages either by using a predefined schema or by using the schema information available in the class (either reflectively or contained in the `SpecificRecord`).
@@ -188,6 +196,7 @@ public static class SinkApplication {
}
----
[[schema-registry-server]]
=== Schema Registry Server
Spring Cloud Stream provides a schema registry server implementation.
@@ -224,6 +233,7 @@ By default, it uses an embedded database.
You can customize the schema storage by using the http://docs.spring.io/spring-boot/docs/current-SNAPSHOT/reference/htmlsingle/#boot-features-sql[Spring Boot SQL database and JDBC configuration options].
[[schema-registry-server-api]]
==== Schema Registry Server API
The Schema Registry Server API consists of the following operations:
@@ -315,6 +325,7 @@ Spring Cloud Stream 1.1.0.RELEASE used the table name, `schema`, for storing `Sc
To avoid any conflicts in the future, starting with 1.1.1.RELEASE, we have opted for the name `SCHEMA_REPOSITORY` for the storage table.
Any Spring Cloud Stream 1.1.0.RELEASE users who upgrade should migrate their existing schemas to the new table before upgrading.
[[using-confluent-s-schema-registry]]
==== Using Confluent's Schema Registry
The default configuration creates a `DefaultSchemaRegistryClient` bean.
@@ -331,6 +342,7 @@ public SchemaRegistryClient schemaRegistryClient(@Value("${spring.cloud.stream.s
----
NOTE: The ConfluentSchemaRegistryClient is tested against Confluent platform version 4.0.0.
[[schema-registration-and-resolution]]
=== Schema Registration and Resolution
To better understand how Spring Cloud Stream registers and resolves new schemas and its use of Avro schema comparison features, we provide two separate subsections:

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

@@ -12,6 +12,7 @@
// ======================================================================================
[[preface]]
== Preface
include::preface.adoc[]
@@ -72,6 +73,7 @@ class BootTestStreamApplicationTests {
}
----
[[main-concepts]]
== Main Concepts
Spring Cloud Stream provides a number of abstractions and primitives that simplify the writing of message-driven microservice applications.
@@ -95,6 +97,7 @@ necessary to establish bindings are handled by middleware-specific _Binder_ impl
.Spring Cloud Stream Application
image::SCSt-with-binder.png[width=800,scaledwidth="75%",align="center"]
[[fat-jar]]
==== Fat JAR
Spring Cloud Stream applications can be run in stand-alone mode from your IDE for testing.
@@ -207,6 +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
To understand the programming model, you should be familiar with the following core concepts:
@@ -217,6 +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 are extension components of Spring Cloud Stream responsible for providing the necessary configuration and implementation to facilitate
@@ -230,6 +235,7 @@ some help in the form of minimalistic yet required set of instructions from the
While it is out of scope of this section to discuss all of the available binder and binding configuration options (the rest of the manual covers them extensively),
_Binding_ as a concept, does require special attention. The next section discusses it in detail.
[[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_.
@@ -263,6 +269,7 @@ And in this context beans of type `Supplier`, `Function` or `Consumer` are treat
binding of to destinations exposed by the provided binder following certain naming conventions and
rules to avoid extra configuration.
[[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,
@@ -276,6 +283,7 @@ The following sub-sections will describe the naming conventions and configuratio
NOTE: If your binding name has special characters, such as the `.` character, you need to surround the binding key with brackets (`[]`) and then wrap it in qoutes.
For example `spring.cloud.stream.bindings."[my.output.binding.key]".destination`.
[[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
@@ -373,6 +381,7 @@ public static class EmptyConfiguration {
As you can see we have declared two input bindings and two output bindings while our configuration had no functions defined, yet we were able to successfully create these bindings and access their corresponding channels.
[[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.
@@ -383,6 +392,7 @@ we recommend using functional implementations.
[[spring_cloud_function]]
==== Spring Cloud Function support
[[overview]]
===== Overview
Since Spring Cloud Stream v2.1, another alternative for defining _stream handlers_ and _sources_ is to use build-in
@@ -449,6 +459,7 @@ public static class SinkFromConsumer {
}
----
[[suppliers-sources]]
===== Suppliers (Sources)
`Function` and `Consumer` are pretty straightforward when it comes to how their invocation is triggered. They are triggered based
@@ -535,6 +546,7 @@ the framework will split the returning sending out each item as an individual me
he desired behavior you can set it to `false` at which point such supplier will simply return
the produced Flux without splitting it.
[[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
@@ -545,6 +557,7 @@ on tracing data stored in thread local.
For those cases we have another mechanism via `StreamBridge`, where user has more control over threading mechanism. You can get more details
in <<Sending arbitrary data to an output (e.g. Foreign event-driven sources)>> section.
[[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.
@@ -570,6 +583,7 @@ The result of such composition is a reactive `Consumer`. However, there is no wa
so this limitation can only be addressed by either making your consumer reactive and subscribing manually (as discussed earlier), or changing your function to be imperative.
[[polling-configuration-properties]]
====== Polling Configuration Properties
The following properties are exposed by Spring Cloud Stream and are prefixed with the `spring.integration.poller.`:
@@ -602,6 +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
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
@@ -610,6 +625,7 @@ you may have several components that require different polling configurations
For such cases please use per-binding way of configuring poller. For example, assume you have an output binding `supply-out-0`. In this case you can configure poller for such
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)
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
@@ -658,6 +674,7 @@ Also, note that `streamBridge.send(..)` method takes an `Object` for data. This
will go through the same routine when sending output as if it was from any Function or Supplier providing the same level
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` can also be used for cases when output destination(s) are not known ahead of time similar to the use cases
@@ -734,11 +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
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
Spring Cloud Stream supports multiple binder scenarios. For example you may be receiving data from Kafka and sending it to RabbitMQ.
@@ -760,6 +779,7 @@ As you can see there is one additional argument that you can provide - `binderTy
NOTE: For cases where `spring.cloud.stream.output-bindings` property is used or the binding was already created under different binder, the `binderType`
argument will have no effect.
[[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`.
@@ -825,6 +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
@@ -863,6 +884,7 @@ Reactive function is _initialization function_. It is invoked only once to get a
Therefore, with reactive functions you must rely on the richness of the reactive API when it comes to error handling and retries (i.e., `doOnError()`, `.onError*()` etc).
====
[[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.
@@ -893,6 +915,7 @@ For example, if we want to give our `toUpperCase|wrapInQuotes` a more descriptiv
with the following property `spring.cloud.stream.function.bindings.toUpperCase|wrapInQuotes-in-0=quotedUpperCaseInput` allowing
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
Function composition effectively allows you to address complexity by breaking it down
@@ -944,6 +967,7 @@ While trivial, this example demonstrates how one function enriches the incoming
so the other function - `echo` - can benefit form it. The `echo` function stays clean and focused on business logic only.
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
Starting with version 3.0 spring-cloud-stream provides support for functions that
@@ -1062,6 +1086,7 @@ public void testSingleInputMultiOutput() {
}
----
[[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
@@ -1119,6 +1144,7 @@ public void testMultipleFunctions() {
}
----
[[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
@@ -1132,6 +1158,7 @@ public Function<List<Person>, Person> findFirstPerson() {
}
----
[[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
@@ -1156,6 +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
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
@@ -1207,6 +1235,7 @@ into various patterns you can apply by merging the best of Spring Integration an
[[spring-cloud-streams-overview-using-polled-consumers]]
==== Using Polled Consumers
[[overview]]
===== Overview
When using polled consumers, you poll the `PollableMessageSource` on demand.
@@ -1325,6 +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, in the context of Spring Cloud Stream, is the ability to either
@@ -1332,6 +1362,7 @@ _a) route events to a particular event subscriber_ or
_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 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.
@@ -1345,6 +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
[source,java]
----
@@ -1381,6 +1413,7 @@ For example,
setting `spring.cloud.function.routing-expression` header to value `T(java.lang.System).currentTimeMillis() % 2 == 0 ? 'even' : 'odd'` will end up semi-randomly routing request to either `odd` or `even` functions.
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
The `spring.cloud.function.routing-expression` and/or `spring.cloud.function.definition`
@@ -1409,6 +1442,7 @@ public class RoutingStreamApplication {
IMPORTANT: Passing instructions via application properties is especially important for reactive functions given that a reactive
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
`RoutingFunction` is a `Function` and as such treated no differently than any other function. Well. . . almost.
@@ -1425,12 +1459,14 @@ becomes as a `Consumer` by not having any output bindings. However, if `RoutingF
the output, the output binding for the `RoutingFunction` will be create dynamically at which point `RoutingFunction` will act as a regular `Function`
with regards to bindings (having both input and output bindings).
[[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
You can also delegate to the framework to dynamically resolve the output destination by specifying `spring.cloud.stream.sendto.destination` header
@@ -1488,6 +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)
Once function is invoked, its result is sent by the framework to a target destination which effectively completes function invocation cycle.
@@ -1600,6 +1637,7 @@ public Function<Flux<String>, Flux<String>> uppercase() {
}
----
[[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
@@ -1607,6 +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
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
@@ -1645,6 +1684,7 @@ NOTE: If by accident you declare such handler as a `Function`, it will still wor
If you want to have a single error handler for all function beans, you can use the standard spring-cloud-stream mechanism for defining default properties `spring.cloud.stream.default.error-handler-definition=myErrorHandler`
[[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_.
@@ -1710,6 +1750,7 @@ You can also facilitate immediate dispatch to DLQ (without re-tries) by setting
--spring.cloud.stream.bindings.uppercase-in-0.consumer.max-attempts=1
----
[[retry-template]]
==== Retry Template
In this section we cover configuration properties relevant to configuration of retry capabilities.
@@ -1776,6 +1817,7 @@ spring.cloud.stream.bindings.<foo>.consumer.retry-template-name=<your-retry-temp
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
The following image shows the general relationship of producers and consumers:
@@ -1831,11 +1873,13 @@ In the https://medium.com/@domenicosibilio/how-to-create-a-spring-cloud-stream-b
in details, with an example, a set of steps necessary to implement a custom binder.
The steps are also highlighted in the `<<spring-cloud-stream-overview-custom-binder-impl>>` section.
[[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
By default, Spring Cloud Stream relies on Spring Boot's auto-configuration to configure the binding process.
@@ -1975,6 +2019,7 @@ Note that, when there are more than one instance of the same type of the binder,
=== 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
Since version 3.1 we expose `org.springframework.cloud.stream.binding.BindingsLifecycleController` which is registered as bean and once
@@ -1994,6 +2039,7 @@ bindingsController.changeState("echo-in-0", State.STOPPED);
assertThat(binding.isRunning()).isFalse();
----
[[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:
@@ -2052,6 +2098,7 @@ You can also stop, start, pause, and resume individual bindings by posting to th
NOTE: `PAUSED` and `RESUMED` work only when the corresponding binder and its underlying technology supports it. Otherwise, you see the warning message in the logs.
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
The following properties are available when customizing binder configurations. These properties exposed via `org.springframework.cloud.stream.config.BinderProperties`
@@ -2326,6 +2373,7 @@ myFileBinder:\
com.example.springcloudstreamcustombinder.config.FileMessageBinderConfiguration
----
[[configuration-options]]
== Configuration Options
Spring Cloud Stream supports general configuration options as well as configuration for bindings and binders.
@@ -2334,6 +2382,7 @@ Some binders let additional binding properties support middleware-specific featu
Configuration options can be provided to Spring Cloud Stream applications through any mechanism supported by Spring Boot.
This includes application arguments, environment variables, and YAML or .properties files.
[[binding-service-properties]]
=== Binding Service Properties
These properties are exposed via `org.springframework.cloud.stream.config.BindingServiceProperties`
@@ -2396,6 +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
These properties are exposed via `org.springframework.cloud.stream.config.BindingProperties`
@@ -2427,6 +2477,7 @@ See `<<multiple-binders>>` for details.
+
Default: `null` (the default binder is used, if it exists).
[[consumer-properties]]
==== Consumer Properties
These properties are exposed via `org.springframework.cloud.stream.binder.ConsumerProperties`
@@ -2513,6 +2564,7 @@ When set to true, the underlying binder will natively multiplex destinations on
+
Default: `false`.
[[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.
@@ -2536,6 +2588,7 @@ public MessageSourceCustomizer<AmqpMessageSource> sourceCustomizer() {
----
====
[[producer-properties]]
==== Producer Properties
These properties are exposed via `org.springframework.cloud.stream.binder.ProducerProperties`
@@ -2606,6 +2659,7 @@ When set to true, if the binder supports asynchroous send results, send failures
Default: false.
[[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
@@ -2650,6 +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
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:
@@ -2695,6 +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
As mentioned earlier, for the framework to select the appropriate `MessageConverter`, it requires argument type and, optionally, content type information.
@@ -2716,6 +2772,7 @@ NOTE: Do not expect `Message` to be converted into some other type based only on
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
`MessageConverters` define two methods:
@@ -2746,6 +2803,7 @@ Object fromMessage(Message<?> message, Class<?> targetClass);
Message<byte[]> toMessage(Object payload, @Nullable MessageHeaders headers);
----
[[provided-messageconverters]]
=== Provided MessageConverters
As mentioned earlier, the framework already provides a stack of `MessageConverters` to handle most common use cases.
@@ -2804,6 +2862,7 @@ public class MyCustomMessageConverter extends AbstractMessageConverter {
}
----
[[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:
@@ -2929,6 +2988,7 @@ 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
Spring Cloud Stream provides support for testing your microservice applications without connecting to a messaging system.
@@ -2941,6 +3001,7 @@ Spring Cloud Stream comes with a test binder which you can use for testing the v
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
To enable Spring Integration test binder, all you need is to add it as a dependency.
@@ -2964,6 +3025,7 @@ Or for build.gradle.kts
testImplementation("org.springframework.cloud:spring-cloud-stream-test-binder")
----
[[test-binder-usage]]
==== Test Binder usage
Now you can test your microservice as a simple unit test
@@ -3106,6 +3168,7 @@ public void testMultipleFunctions() {
----
[[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).
@@ -3188,6 +3251,7 @@ Message 3: MY OWN DATA 20BF2E64-7FF4-4CB6-A823-4053D30B5C74
NOTE: DO NOT name this bean `messageSource` as it is going to be in conflict with the bean of the same name (different type)
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
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.
@@ -3196,6 +3260,7 @@ When the test binder is present on the test classpath, Spring Cloud Stream will
In other words, you cannot mix both the test binder and a regular middleware binder for testing purposes in the same module.
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
Spring Cloud Stream provides a health indicator for binders.
@@ -3229,16 +3294,19 @@ Since Kafka binder is not used and it has specific checks to see if any destinat
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
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
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
The following is the list of available binder implementations