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

Migrate Structure

Browse Source
This commit is contained in:
Marcin Grzejszczak
2023-09-08 16:12:37 +02:00
committed by Oleg Zhurakousky
parent 0c256343c7
commit 40fd8595d4
34 changed files with 0 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
docs/modules/ROOT/assets/images/SCSt-groups.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 17 KiB

BIN
docs/modules/ROOT/assets/images/SCSt-overview.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 512 KiB

BIN
docs/modules/ROOT/assets/images/SCSt-partitioning.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 18 KiB

BIN
docs/modules/ROOT/assets/images/SCSt-sensors.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 16 KiB

BIN
docs/modules/ROOT/assets/images/SCSt-with-binder.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 18 KiB

BIN
docs/modules/ROOT/assets/images/custom_vs_global_error_channels.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 47 KiB

BIN
docs/modules/ROOT/assets/images/producers-consumers.png Executable file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 16 KiB

BIN
docs/modules/ROOT/assets/images/redis-binder.png Executable file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 13 KiB

BIN
docs/modules/ROOT/assets/images/registration.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 22 KiB

BIN
docs/modules/ROOT/assets/images/schema_reading.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 45 KiB

BIN
docs/modules/ROOT/assets/images/schema_resolution.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 27 KiB

BIN
docs/modules/ROOT/assets/images/spring-initializr.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 311 KiB

26
docs/modules/ROOT/pages/README.adoc Normal file
View File

@@ -0,0 +1,26 @@
:jdkversion: 17
:github-tag: master
:github-repo: spring-cloud/spring-cloud-stream
:github-raw: https://raw.githubusercontent.com/{github-repo}/{github-tag}
:github-code: https://github.com/{github-repo}/tree/{github-tag}
image::https://circleci.com/gh/spring-cloud/spring-cloud-stream.svg?style=svg["CircleCI", link="https://circleci.com/gh/spring-cloud/spring-cloud-stream"]
image::https://codecov.io/gh/spring-cloud/spring-cloud-stream/branch/{github-tag}/graph/badge.svg["codecov", link="https://codecov.io/gh/spring-cloud/spring-cloud-stream"]
// ======================================================================================
== Introduction
include::intro.adoc[]
== Resources
For more information, please visit the https://spring.io/projects/spring-cloud-stream[project website]:
== Building
include::https://raw.githubusercontent.com/spring-cloud/spring-cloud-build/master/docs/src/main/asciidoc/building.adoc[]
== Contributing
include::https://raw.githubusercontent.com/spring-cloud/spring-cloud-build/master/docs/src/main/asciidoc/contributing.adoc[]

22
docs/modules/ROOT/pages/_configprops.adoc Normal file
View File

@@ -0,0 +1,22 @@
|===
|Name | Default | Description
|spring.cloud.stream.binders | | Additional per-binder properties (see {@link BinderProperties}) if more then one binder of the same type is used (i.e., connect to multiple instances of RabbitMq). Here you can specify multiple binder configurations, each with different environment settings. For example; spring.cloud.stream.binders.rabbit1.environment. . . , spring.cloud.stream.binders.rabbit2.environment. . .
|spring.cloud.stream.binding-retry-interval | `+++30+++` | Retry interval (in seconds) used to schedule binding attempts. Default: 30 sec.
|spring.cloud.stream.bindings | | Additional binding properties (see {@link BinderProperties}) per binding name (e.g., 'input`). For example; This sets the content-type for the 'input' binding of a Sink application: 'spring.cloud.stream.bindings.input.contentType=text/plain'
|spring.cloud.stream.default-binder | | The name of the binder to use by all bindings in the event multiple binders available (e.g., 'rabbit').
|spring.cloud.stream.dynamic-destination-cache-size | `+++10+++` | The maximum size of Least Recently Used (LRU) cache of dynamic destinations. Once this size is reached, new destinations will trigger the removal of old destinations. Default: 10
|spring.cloud.stream.dynamic-destinations | `+++[]+++` | A list of destinations that can be bound dynamically. If set, only listed destinations can be bound.
|spring.cloud.stream.function.bindings | |
|spring.cloud.stream.input-bindings | | A semi-colon delimited string to explicitly define input bindings (specifically for cases when there is no implicit trigger to create such bindings such as Function, Supplier or Consumer).
|spring.cloud.stream.instance-count | `+++1+++` | The number of deployed instances of an application. Default: 1. NOTE: Could also be managed per individual binding "spring.cloud.stream.bindings.foo.consumer.instance-count" where 'foo' is the name of the binding.
|spring.cloud.stream.instance-index | `+++0+++` | The instance id of the application: a number from 0 to instanceCount-1. Used for partitioning and with Kafka. NOTE: Could also be managed per individual binding "spring.cloud.stream.bindings.foo.consumer.instance-index" where 'foo' is the name of the binding.
|spring.cloud.stream.instance-index-list | | A list of instance id's from 0 to instanceCount-1. Used for partitioning and with Kafka. NOTE: Could also be managed per individual binding "spring.cloud.stream.bindings.foo.consumer.instance-index-list" where 'foo' is the name of the binding. This setting will override the one set in 'spring.cloud.stream.instance-index'
|spring.cloud.stream.integration.message-handler-not-propagated-headers | | Message header names that will NOT be copied from the inbound message.
|spring.cloud.stream.output-bindings | | A semi-colon delimited string to explicitly define output bindings (specifically for cases when there is no implicit trigger to create such bindings such as Function, Supplier or Consumer).
|spring.cloud.stream.override-cloud-connectors | `+++false+++` | This property is only applicable when the cloud profile is active and Spring Cloud Connectors are provided with the application. If the property is false (the default), the binder detects a suitable bound service (for example, a RabbitMQ service bound in Cloud Foundry for the RabbitMQ binder) and uses it for creating connections (usually through Spring Cloud Connectors). When set to true, this property instructs binders to completely ignore the bound services and rely on Spring Boot properties (for example, relying on the spring.rabbitmq.* properties provided in the environment for the RabbitMQ binder). The typical usage of this property is to be nested in a customized environment when connecting to multiple systems.
|spring.cloud.stream.pollable-source | `+++none+++` | A semi-colon delimited list of binding names of pollable sources. Binding names follow the same naming convention as functions. For example, name '...pollable-source=foobar' will be accessible as 'foobar-iin-0'' binding
|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}
|===

15
docs/modules/ROOT/pages/binders.adoc Normal file
View File

@@ -0,0 +1,15 @@
*{project-version}*
[[binders]]
== Binder Implementations
The following binder implementations are available:
* https://cloud.spring.io/spring-cloud-stream-binder-rabbit/[RabbitMQ]
* https://cloud.spring.io/spring-cloud-stream-binder-kafka/[Apache Kafka]
* https://github.com/spring-cloud/spring-cloud-stream-binder-aws-kinesis[Amazon Kinesis]
* https://github.com/spring-cloud/spring-cloud-gcp/tree/master/spring-cloud-gcp-pubsub-stream-binder[Google PubSub _(partner maintained)_]
* https://github.com/SolaceProducts/solace-spring-cloud/tree/master/solace-spring-cloud-starters/solace-spring-cloud-stream-starter#spring-cloud-stream-binder-for-solace-pubsub[Solace PubSub+ _(partner maintained)_]
* https://aka.ms/spring/docs#spring-cloud-stream-binder-for-azure-event-hubs[Azure Event Hubs _(partner maintained)_]
* https://aka.ms/spring/docs#spring-cloud-stream-binder-for-azure-service-bus[Azure Service Bus Binder _(partner maintained)_]
* https://github.com/alibaba/spring-cloud-alibaba/wiki/RocketMQ-en[Apache RocketMQ _(partner maintained)_]

29
docs/modules/ROOT/pages/index.adoc Normal file
View File

@@ -0,0 +1,29 @@
= 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
*{project-version}*
:docinfo: shared
The reference documentation consists of the following sections:
[horizontal]
<<spring-cloud-stream.adoc#spring-cloud-stream-reference,Overview>> :: History, Quick Start, Concepts, Architecture Overview, Binder Abstraction, and Core Features
<<spring-cloud-stream-binder-rabbit#spring-cloud-stream-binder-rabbit-reference,Rabbit MQ Binder>> :: Spring Cloud Stream binder reference for Rabbit MQ
<<spring-cloud-stream-binder-kafka#_apache_kafka_binder,Apache Kafka Binder>> :: Spring Cloud Stream binder reference for Apache Kafka
<<spring-cloud-stream-binder-kafka#_kafka_streams_binder,Apache Kafka Streams binder>> :: Spring Cloud Stream binder reference for Apache Kafka Streams
<<spring-cloud-stream-binder-pulsar#spring-cloud-stream-binder-pulsar-reference,Apache Pulsar Binder>> :: Spring Cloud Stream binder reference for Apache Pulsar
https://github.com/SolaceProducts/solace-spring-cloud/tree/master/solace-spring-cloud-starters/solace-spring-cloud-stream-starter#spring-cloud-stream-binder-for-solace-pubsub[Solace PubSub+ Binder] :: Spring Cloud Stream binder reference for Solace PubSub+ (Partner Maintained)
<<binders.adoc#binders,Additional Binders>> :: A collection of Partner maintained binder implementations for Spring Cloud Stream (e.g., Azure Event Hubs, Google PubSub, Solace PubSub+)
<<spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-schema-registry-reference,Schema Registry>> :: Spring Cloud Stream Schema Registry Reference
https://github.com/spring-cloud/spring-cloud-stream-samples/[Spring Cloud Stream Samples] :: A curated collection of repeatable Spring Cloud Stream samples to walk through the features
Relevant Links:
[horizontal]
https://cloud.spring.io/spring-cloud-dataflow/[Spring Cloud Data Flow] :: Spring Cloud Data Flow
http://www.enterpriseintegrationpatterns.com/[Enterprise Integration Patterns] :: Patterns and Best Practices for Enterprise Integration
https://spring.io/projects/spring-integration[Spring Integration] :: Spring Integration framework

69
docs/modules/ROOT/pages/intro.adoc Normal file
View File

@@ -0,0 +1,69 @@
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.
It provides opinionated configuration of middleware from several vendors, introducing the concepts of persistent publish-subscribe semantics, consumer groups, and partitions.
These are called binder implementations in the parlance of Spring Cloud Stream.
Out of the box, Spring Cloud Stream provides binder implementations for Apache Kafka and RabbitMQ.
While these two binder implementations are based on Message Channels, Spring Cloud Stream also provides another binder implementation for Kafka Streams that does not use message channels, but native Kafka Streams types such as KStream, KTable etc.
Below, you can find more information on how to use these various out-of-the-box binder implementations in your applications.
== Apache Kafka Binder
=== 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:
[source,xml]
----
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>
----
Alternatively, you can also use the Spring Cloud Stream Kafka Starter, as shown in the following example for Maven:
[source,xml]
----
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-kafka</artifactId>
</dependency>
----
== Apache Kafka Streams Binder
=== 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:
[source,xml]
----
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka-streams</artifactId>
</dependency>
----
== RabbitMQ Binder
=== Usage
To use the RabbitMQ binder, you can add it to your Spring Cloud Stream application, by using the following Maven coordinates:
[source,xml]
----
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-rabbit</artifactId>
</dependency>
----
Alternatively, you can use the Spring Cloud Stream RabbitMQ Starter, as follows:
[source,xml]
----
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
----

229
docs/modules/ROOT/pages/kafka/kafka-reactive.adoc Normal file
View File

@@ -0,0 +1,229 @@
== 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
Following are the maven coordinates for the reactive Kafka binder.
```
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka-reactive</artifactId>
</dependency>
```
=== 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.
====
[source, java]
----
@Bean
public Function<Flux<String>, Flux<String>> uppercase() {
return s -> s.map(String::toUpperCase);
}
----
====
You can use the above `upppercase` function with both message channel based Kafka binder (`spring-cloud-stream-binder-kafka`) as well as the reactive Kafka binder (`spring-cloud-stream-binder-kafka-reactive`), the topic of discussion in this section.
When using this function with the regular Kafka binder, although you are using reactive types in the application (i.e., in the `uppercase` function), you only get the reactive streams within the execution of your function.
Outside the function's execution context, there is no reactive benefits since the underlying binder is not based on the reactive stack.
Therefore, although this might look like it is bringing a full end-to-end reactive stack, this application is only partially reactive.
Now assume that you are using the proper reactive binder for Kafka - `spring-cloud-stream-binder-kafka-reactive` with the above function's application.
This binder implementation will give the full reactive benefits all the way from consumption on the top end to publishing at the bottom end of the chain.
This is because the underlying binder is built on top of https://projectreactor.io/docs/kafka/release/reference/[Reactor Kafka]'s core API's.
On the consumer side, it makes use of the https://projectreactor.io/docs/kafka/release/reference/#api-guide-receiver[KafkaReceiver] which is a reactive implementation of a Kafka consumer.
Similarly, on the producer side, it uses https://projectreactor.io/docs/kafka/release/reference/#api-guide-sender[KafkaSender] API which is the reactive implementation of a Kafka producer.
Since the foundations of the reactive Kafka binder is built upon a proper reactive Kafka API, applications get the full benefits of using reactive technologies.
Things like automatic back pressure, among other reactive capabilities, are built-in for the application when using this reactive Kafka binder.
Starting with version 4.0.2, you can customize the `ReceiverOptions` and `SenderOptions` by providing one or more `ReceiverOptionsCustomizer` or `SenderOptionsCustomizer` beans respectively.
They are `BiFunction` s which receive the binding name and initial options, returning the customized options.
The interfaces extend `Ordered` so the customizers will be applied in the order required, when more than one are present.
IMPORTANT: The binder does not commit offsets by default.
Starting with version 4.0.2, the `KafkaHeaders.ACKNOWLEDGMENT` header contains a `ReceiverOffset` object which allows you to cause the offset to be committed by calling its `acknowledge()` or `commit()` methods.
====
[source, java]
----
@Bean
public Consumer<Flux<Message<String>> consume() {
return msg -> {
process(msg.getPayload());
msg.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, ReceiverOffset.class).acknowledge();
}
}
----
====
Refer to the `reactor-kafka` documentation and javadocs for more information.
In addition, starting with version 4.0.3, the Kafka consumer property `reactiveAtmostOnce` can be set to `true` and the binder will automatically commit the offsets before records returned by each poll are processed.
Also, starting with version 4.0.3, you can set the consumer property `reactiveAutoCommit` to `true` and the the binder will automatically commit the offsets after the records returned by each poll are processed.
In these cases, the acknowledgment header is not present.
IMPORTANT: 4.0.2 also provided `reactiveAutoCommit`, but the implementation was incorrect, it behaved similarly to `reactiveAtMostOnce`.
The following is an example of how to use `reaciveAutoCommit`.
====
[source, java]
----
@Bean
Consumer<Flux<Flux<ConsumerRecord<?, String>>>> input() {
return flux -> flux
.doOnNext(inner -> inner
.doOnNext(val -> {
log.info(val.value());
})
.subscribe())
.subscribe();
}
----
====
Note that `reactor-kafka` returns a `Flux<Flux<ConsumerRecord<?, ?>>>` when using auto commit.
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
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`.
Here is such a function.
====
[source, java]
----
@Bean
public Function<Flux<ReceiverRecord<byte[], byte[]>>, Flux<String>> lowercase() {
return s -> s.map(rec -> new String(rec.value()).toLowerCase());
}
----
====
In this function, note that, we are consuming the record as `Flux<ReceiverRecord<byte[], byte[]>>` and then producing it as `Flux<String>`.
`ReceiverRecord` is the basic received record which is a specialized Kafka `ConsumerRecord` in Reactor Kafka.
When using the reactive Kafka binder, the above function will give you access to the `ReceiverRecord` type for each incoming record.
However, in this case, you need to provide a custom implementation for a https://docs.spring.io/spring-kafka/docs/current/api/org/springframework/kafka/support/converter/RecordMessageConverter.html[RecordMessageConverter].
By default, the reactive Kafka binder uses a https://docs.spring.io/spring-kafka/docs/current/api/org/springframework/kafka/support/converter/MessagingMessageConverter.html[MessagingMessageConverter] that converts the payload and headers from the `ConsumerRecord`.
Therefore, by the time your handler method receives it, the payload is already extracted from the received record and passed onto the method as in the case of the first function we looked above.
By providing a custom `RecordMessageConverter` implementation in the application, you can override the default behavior.
For example, if you want to consume the record as raw `Flux<ReceiverRecord<byte[], byte[]>>`, then you can provide the following bean definition in the application.
====
[source, java]
----
@Bean
RecordMessageConverter fullRawReceivedRecord() {
return new RecordMessageConverter() {
private final RecordMessageConverter converter = new MessagingMessageConverter();
@Override
public Message<?> toMessage(ConsumerRecord<?, ?> record, Acknowledgment acknowledgment,
Consumer<?, ?> consumer, Type payloadType) {
return MessageBuilder.withPayload(record).build();
}
@Override
public ProducerRecord<?, ?> fromMessage(Message<?> message, String defaultTopic) {
return this.converter.fromMessage(message, defaultTopic);
}
};
}
----
====
Then, you need to instruct the framework to use this converter for the required binding.
Here is an example based on our `lowercase` function.
```
spring.cloud.stream.kafka.bindings.lowercase-in-0.consumer.converterBeanName=fullRawReceivedRecord"
```
`lowercase-in-0` is the input binding name for our `lowercase` function.
For the outbound (`lowecase-out-0`), we still use the regular `MessagingMessageConverter`.
In the `toMessage` implementation above, we receive the raw `ConsumerRecord` (`ReceiverRecord` since we are in a reactive binder context) and then wrap it inside a `Message`.
Then that message payload which is the `ReceiverRecord` is provided to the user method.
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
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.
This could be useful when you are consuming records from a partitioned topic.
For example, assume that the incoming topic has at least three partitions.
Then you can set the following property.
```
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
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
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
Starting with version 4.0.3, you can configure the `resultMetadataChannel` to receive `SenderResult<?>` s to determine success/failure of sends.
The `SenderResult` contains `correlationMetadata` to allow you to correlate results with sends; it also contains `RecordMetadata`, which indicates the `TopicPartition` and offset of the sent record.
The `resultMetadataChannel` **must** be a `FluxMessageChannel` instance.
Here is an example of how to use this feature, with correlation metadata of type `Integer`:
====
[source, java]
----
@Bean
FluxMessageChannel sendResults() {
return new FluxMessageChannel();
}
@ServiceActivator(inputChannel = "sendResults")
void handleResults(SenderResult<Integer> result) {
if (result.exception() != null) {
failureFor(result);
}
else {
successFor(result);
}
}
----
====
To set the correlation metadata on an output record, set the `CORRELATION_ID` header:
====
[source, java]
----
streamBridge.send("words1", MessageBuilder.withPayload("foobar")
.setCorrelationId(42)
.build());
----
====
When using the feature with a `Function`, the function output type must be a `Message<?>` with the correlation id header set to the desired value.
Metadata should be unique, at least for the duration of the send.

2096
docs/modules/ROOT/pages/kafka/kafka-streams.adoc Normal file
View File

File diff suppressed because it is too large Load Diff

150
docs/modules/ROOT/pages/kafka/kafka_dlq.adoc Normal file
View File

@@ -0,0 +1,150 @@
[[kafka-dlq-processing]]
=== Dead-Letter Topic Processing
[[dlq-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.
To change this behavior, add a `DlqPartitionFunction` implementation as a `@Bean` to the application context.
Only one such bean can be present.
The function is provided with the consumer group, the failed `ConsumerRecord` and the exception.
For example, if you always want to route to partition 0, you might use:
====
[source, java]
----
@Bean
public DlqPartitionFunction partitionFunction() {
return (group, record, ex) -> 0;
}
----
====
NOTE: If you set a consumer binding's `dlqPartitions` property to 1 (and the binder's `minPartitionCount` is equal to `1`), there is no need to supply a `DlqPartitionFunction`; the framework will always use partition 0.
If you set a consumer binding's `dlqPartitions` property to a value greater than `1` (or the binder's `minPartitionCount` is greater than `1`), you **must** provide a `DlqPartitionFunction` bean, even if the partition count is the same as the original topic's.
It is also possible to define a custom name for the DLQ topic.
In order to do so, create an implementation of `DlqDestinationResolver` as a `@Bean` to the application context.
When the binder detects such a bean, that takes precedence, otherwise it will use the `dlqName` property.
If neither of these are found, it will default to `error.<destination>.<group>`.
Here is an example of `DlqDestinationResolver` as a `@Bean`.
====
[source]
----
@Bean
public DlqDestinationResolver dlqDestinationResolver() {
return (rec, ex) -> {
if (rec.topic().equals("word1")) {
return "topic1-dlq";
}
else {
return "topic2-dlq";
}
};
}
----
====
One important thing to keep in mind when providing an implementation for `DlqDestinationResolver` is that the provisioner in the binder will not auto create topics for the application.
This is because there is no way for the binder to infer the names of all the DLQ topics the implementation might send to.
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
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.
However, if the problem is a permanent issue, that could cause an infinite loop.
The sample Spring Boot application within this topic is an example of how to route those messages back to the original topic, but it moves them to a "`parking lot`" topic after three attempts.
The application is another spring-cloud-stream application that reads from the dead-letter topic.
It exits when no messages are received for 5 seconds.
The examples assume the original destination is `so8400out` and the consumer group is `so8400`.
There are a couple of strategies to consider:
* Consider running the rerouting only when the main application is not running.
Otherwise, the retries for transient errors are used up very quickly.
* Alternatively, use a two-stage approach: Use this application to route to a third topic and another to route from there back to the main topic.
The following code listings show the sample application:
.application.properties
[source]
----
spring.cloud.stream.bindings.input.group=so8400replay
spring.cloud.stream.bindings.input.destination=error.so8400out.so8400
spring.cloud.stream.bindings.output.destination=so8400out
spring.cloud.stream.bindings.parkingLot.destination=so8400in.parkingLot
spring.cloud.stream.kafka.binder.configuration.auto.offset.reset=earliest
spring.cloud.stream.kafka.binder.headers=x-retries
----
.Application
[source,java]
----
@SpringBootApplication
public class ReRouteDlqKApplication implements CommandLineRunner {
private static final String X_RETRIES_HEADER = "x-retries";
public static void main(String[] args) {
SpringApplication.run(ReRouteDlqKApplication.class, args).close();
}
private final AtomicInteger processed = new AtomicInteger();
@Autowired
private StreamBridge streamBridge;
@Bean
public Function<Message<?>, Message<?>> reRoute() {
return failed -> {
processed.incrementAndGet();
Integer retries = failed.getHeaders().get(X_RETRIES_HEADER, Integer.class);
if (retries == null) {
System.out.println("First retry for " + failed);
return MessageBuilder.fromMessage(failed)
.setHeader(X_RETRIES_HEADER, 1)
.setHeader(BinderHeaders.PARTITION_OVERRIDE,
failed.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID))
.build();
}
else if (retries < 3) {
System.out.println("Another retry for " + failed);
return MessageBuilder.fromMessage(failed)
.setHeader(X_RETRIES_HEADER, retries + 1)
.setHeader(BinderHeaders.PARTITION_OVERRIDE,
failed.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID))
.build();
}
else {
System.out.println("Retries exhausted for " + failed);
streamBridge.send("parkingLot", MessageBuilder.fromMessage(failed)
.setHeader(BinderHeaders.PARTITION_OVERRIDE,
failed.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID))
.build());
}
return null;
};
}
@Override
public void run(String... args) throws Exception {
while (true) {
int count = this.processed.get();
Thread.sleep(5000);
if (count == this.processed.get()) {
System.out.println("Idle, exiting");
return;
}
}
}
}
----

1075
docs/modules/ROOT/pages/kafka/kafka_overview.adoc Normal file
View File

File diff suppressed because it is too large Load Diff

121
docs/modules/ROOT/pages/kafka/kafka_partitions.adoc Normal file
View File

@@ -0,0 +1,121 @@
=== Partitioning with the Kafka Binder
Apache Kafka supports topic partitioning natively.
Sometimes it is advantageous to send data to specific partitions -- for example, when you want to strictly order message processing (all messages for a particular customer should go to the same partition).
The following example shows how to configure the producer and consumer side:
[source,java]
----
@SpringBootApplication
public class KafkaPartitionProducerApplication {
private static final Random RANDOM = new Random(System.currentTimeMillis());
private static final String[] data = new String[] {
"foo1", "bar1", "qux1",
"foo2", "bar2", "qux2",
"foo3", "bar3", "qux3",
"foo4", "bar4", "qux4",
};
public static void main(String[] args) {
new SpringApplicationBuilder(KafkaPartitionProducerApplication.class)
.web(false)
.run(args);
}
@Bean
public Supplier<Message<?>> generate() {
return () -> {
String value = data[RANDOM.nextInt(data.length)];
System.out.println("Sending: " + value);
return MessageBuilder.withPayload(value)
.setHeader("partitionKey", value)
.build();
};
}
}
----
.application.yml
[source, yaml]
----
spring:
cloud:
stream:
bindings:
generate-out-0:
destination: partitioned.topic
producer:
partition-key-expression: headers['partitionKey']
partition-count: 12
----
IMPORTANT: It is important to keep in mind that, since Apache Kafka supports partitioning natively, there is no need to rely on binder partitioning as described above unless you are using custom partition keys as in the example or an expression that involves the payload itself.
The binder-provided partitioning selection is otherwise intended for middleware technologies that do not support native partitioning.
Note that we are using a custom key called `partitionKey` in the above example, that will be the determining factor for the partition, thus in this case it is appropriate to use binder partitioning.
When using native Kafka partitioning, i.e, when you do not provide the `partition-key-expression`, then Apache Kafka will select a partition, which by default will be the hash value of the record key over the available number of partitions.
To add a key to an outbound record, set the `KafkaHeaders.KEY` header to the desired key value in a spring-messaging `Message<?>`.
By default, when no record key is provided, Apache Kafka will choose a partition based on the logic described in the https://kafka.apache.org/documentation/#producerconfigs_partitioner.class[Apache Kafka Documentation].
NOTE: The topic must be provisioned to have enough partitions to achieve the desired concurrency for all consumer groups.
The above configuration supports up to 12 consumer instances (6 if their `concurrency` is 2, 4 if their concurrency is 3, and so on).
It is generally best to "`over-provision`" the partitions to allow for future increases in consumers or concurrency.
NOTE: The preceding configuration uses the default partitioning (`key.hashCode() % partitionCount`).
This may or may not provide a suitably balanced algorithm, depending on the key values. In particular, note that this partitioning strategy differs from the default used by a standalone Kafka producer - such as the one used by Kafka Streams, meaning that the same key value may balance differently across partitions when produced by those clients.
You can override this default by using the `partitionSelectorExpression` or `partitionSelectorClass` properties.
Since partitions are natively handled by Kafka, no special configuration is needed on the consumer side.
Kafka allocates partitions across the instances.
NOTE: The partitionCount for a kafka topic may change during runtime (e.g. due to an adminstration task).
The calculated partitions will be different after that (e.g. new partitions will be used then).
Since 4.0.3 of Spring Cloud Stream runtime changes of partition count will be supported.
See also parameter 'spring.kafka.producer.properties.metadata.max.age.ms' to configure update interval.
Due to some limitations it is not possible to use a 'partition-key-expression' which references the 'payload' of a message, the mechanism will be disabled in that case.
The overall behavior is disabled by default and can be enabled using configuration parameter 'producer.dynamicPartitionUpdatesEnabled=true'.
The following Spring Boot application listens to a Kafka stream and prints (to the console) the partition ID to which each message goes:
[source,java]
----
@SpringBootApplication
public class KafkaPartitionConsumerApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(KafkaPartitionConsumerApplication.class)
.web(WebApplicationType.NONE)
.run(args);
}
@Bean
public Consumer<Message<String>> listen() {
return message -> {
int partition = (int) message.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION);
System.out.println(message + " received from partition " + partition);
};
}
}
----
.application.yml
[source, yaml]
----
spring:
cloud:
stream:
bindings:
listen-in-0:
destination: partitioned.topic
group: myGroup
----
You can add instances as needed.
Kafka rebalances the partition allocations.
If the instance count (or `instance count * concurrency`) exceeds the number of partitions, some consumers are idle.

865
docs/modules/ROOT/pages/kafka/kafka_tips.adoc Normal file
View File

@@ -0,0 +1,865 @@
== Tips, Tricks and Recipes
=== Simple DLQ with Kafka
==== 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
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.
```
@Bean
public Consumer<byte[]> processData() {
return s -> {
throw new RuntimeException();
};
}
```
This is a very trivial function that throws an exception for all the records that it processes, but you can take this function and extend it to any other similar situations.
In order to send the records in error to a DLT, we need to provide the following configuration.
```
spring.cloud.stream:
bindings:
processData-in-0:
group: my-group
destination: input-topic
kafka:
bindings:
processData-in-0:
consumer:
enableDlq: true
dlqName: input-topic-dlq
```
In order to activate DLQ, the application must provide a group name.
Anonymous consumers cannot use the DLQ facilities.
We also need to enable DLQ by setting the `enableDLQ` property on the Kafka consumer binding to `true`.
Finally, we can optionally provide the DLT name by providing the `dlqName` on Kafka consumer binding, which otherwise default to `error.input-topic.my-group` in this case.
Note that in the example consumer provided above, the type of the payload is `byte[]`.
By default, the DLQ producer in Kafka binder expects the payload of type `byte[]`.
If that is not the case, then we need to provide the configuration for proper serializer.
For example, let us re-write the consumer function as below:
```
@Bean
public Consumer<String> processData() {
return s -> {
throw new RuntimeException();
};
}
```
Now, we need to tell Spring Cloud Stream, how we want to serialize the data when writing to the DLT.
Here is the modified configuration for this scenario:
```
spring.cloud.stream:
bindings:
processData-in-0:
group: my-group
destination: input-topic
kafka:
bindings:
processData-in-0:
consumer:
enableDlq: true
dlqName: input-topic-dlq
dlqProducerProperties:
configuration:
value.serializer: org.apache.kafka.common.serialization.StringSerializer
```
=== DLQ with Advanced Retry Options
==== Problem Statement
This is similar to the recipe above, but as a developer I would like to configure the way retries are handled.
==== 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.
By default, the binder retires for a maximum of 3 attempts with a one second initial delay, 2.0 multiplier with each back off with a max delay of 10 seconds.
You can change all these configurations as below:
```
spring.cloud.stream.bindings.processData-in-0.consumer.maxAtttempts
spring.cloud.stream.bindings.processData-in-0.consumer.backOffInitialInterval
spring.cloud.stream.bindings.processData-in-0.consumer.backOffMultipler
spring.cloud.stream.bindings.processData-in-0.consumer.backOffMaxInterval
```
If you want, you can also provide a list of retryable exceptions by providing a map of boolean values.
For example,
```
spring.cloud.stream.bindings.processData-in-0.consumer.retryableExceptions.java.lang.IllegalStateException=true
spring.cloud.stream.bindings.processData-in-0.consumer.retryableExceptions.java.lang.IllegalArgumentException=false
```
By default, any exceptions not listed in the map above will be retried.
If that is not desired, then you can disable that by providing,
```
spring.cloud.stream.bindings.processData-in-0.consumer.defaultRetryable=false
```
You can also provide your own `RetryTemplate` and mark it as `@StreamRetryTemplate` which will be scanned and used by the binder.
This is useful when you want more sophisticated retry strategies and policies.
If you have multiple `@StreamRetryTemplate` beans, then you can specify which one your binding wants by using the property,
```
spring.cloud.stream.bindings.processData-in-0.consumer.retry-template-name=<your-retry-template-bean-name>
```
=== Handling Deserialization errors with DLQ
==== 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
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.
Spring for Apache Kafka project offers some great ways to help the binder with this situation.
Let us explore those.
Assuming this is our function:
```
@Bean
public Consumer<String> functionName() {
return s -> {
System.out.println(s);
};
}
```
It is a trivial function that takes a `String` parameter.
We want to bypass the message converters provided by Spring Cloud Stream and want to use native deserializers instead.
In the case of `String` types, it does not make much sense, but for more complex types like AVRO etc. you have to rely on external deserializers and therefore want to delegate the conversion to Kafka.
Now when the consumer receives the data, let us assume that there is a bad record that causes a deserialization error, maybe someone passed an `Integer` instead of a `String` for example.
In that case, if you don't do something in the application, the exception will be propagated through the chain and your application will exit eventually.
In order to handle this, you can add a `ListenerContainerCustomizer` `@Bean` that configures a `DefaultErrorHandler`.
This `DefaultErrorHandler` is configured with a `DeadLetterPublishingRecoverer`.
We also need to configure an `ErrorHandlingDeserializer` for the consumer.
That sounds like a lot of complex things, but in reality, it boils down to these 3 beans in this case.
```
@Bean
public ListenerContainerCustomizer<AbstractMessageListenerContainer<byte[], byte[]>> customizer(DefaultErrorHandler errorHandler) {
return (container, dest, group) -> {
container.setCommonErrorHandler(errorHandler);
};
}
```
```
@Bean
public DefaultErrorHandler errorHandler(DeadLetterPublishingRecoverer deadLetterPublishingRecoverer) {
return new DefaultErrorHandler(deadLetterPublishingRecoverer);
}
```
```
@Bean
public DeadLetterPublishingRecoverer publisher(KafkaOperations bytesTemplate) {
return new DeadLetterPublishingRecoverer(bytesTemplate);
}
```
Let us analyze each of them.
The first one is the `ListenerContainerCustomizer` bean that takes a `DefaultErrorHandler`.
The container is now customized with that particular error handler.
You can learn more about container customization https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/spring-cloud-stream.html#_advanced_consumer_configuration[here].
The second bean is the `DefaultErrorHandler` that is configured with a publishing to a `DLT`.
See https://docs.spring.io/spring-kafka/docs/current/reference/html/#seek-to-current[here] for more details on `DefaultErrorHandler`.
The third bean is the `DeadLetterPublishingRecoverer` that is ultimately responsible for sending to the `DLT`.
By default, the `DLT` topic is named as the ORIGINAL_TOPIC_NAME.DLT.
You can change that though.
See the https://docs.spring.io/spring-kafka/docs/current/reference/html/#dead-letters[docs] for more details.
We also need to configure an https://docs.spring.io/spring-kafka/docs/current/reference/html/#error-handling-deserializer[ErrorHandlingDeserializer] through application config.
The `ErrorHandlingDeserializer` delegates to the actual deserializer.
In case of errors, it sets key/value of the record to be null and includes the raw bytes of the message.
It then sets the exception in a header and passes this record to the listener, which then calls the registered error handler.
Following is the configuration required:
```
spring.cloud.stream:
function:
definition: functionName
bindings:
functionName-in-0:
group: group-name
destination: input-topic
consumer:
use-native-decoding: true
kafka:
bindings:
functionName-in-0:
consumer:
enableDlq: true
dlqName: dlq-topic
dlqProducerProperties:
configuration:
value.serializer: org.apache.kafka.common.serialization.StringSerializer
configuration:
value.deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
spring.deserializer.value.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
```
We are providing the `ErrorHandlingDeserializer` through the `configuration` property on the binding.
We are also indicating that the actual deserializer to delegate is the `StringDeserializer`.
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
==== 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
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.
Here is it in a gist:
Kafka supports two types of offsets to start with by default - `earliest` and `latest`.
Their semantics are self-explanatory from their names.
Assuming you are running the consumer for the first time.
If you miss the group.id in your Spring Cloud Stream application, then it becomes an anonymous consumer.
Whenever, you have an anonymous consumer, in that case, Spring Cloud Stream application by default will start from the `latest` available offset in the topic partition.
On the other hand, if you explicitly specify a group.id, then by default, the Spring Cloud Stream application will start from the `earliest` available offset in the topic partition.
In both cases above (consumers with explicit groups and anonymous groups), the starting offset can be switched around by using the property `spring.cloud.stream.kafka.bindings.<binding-name>.consumer.startOffset` and setting it to either `earliest` or `latest`.
Now, assume that you already ran the consumer before and now starting it again.
In this case, the starting offset semantics in the above case do not apply as the consumer finds an already committed offset for the consumer group (In the case of an anonymous consumer, although the application does not provide a group.id, the binder will auto generate one for you).
It simply picks up from the last committed offset onward.
This is true, even when you have a `startOffset` value provided.
However, you can override the default behavior where the consumer starts from the last committed offset by using the `resetOffsets` property.
In order to do that, set the property `spring.cloud.stream.kafka.bindings.<binding-name>.consumer.resetOffsets` to `true` (which is `false` by default).
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
==== 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
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.
However, there are some low-level strategies that the binder provides to achieve this use case.
Let's explore them.
First of all, when you want to reset to an arbitrary offset other than `earliest` or `latest`, make sure to leave the `resetOffsets` configuration to its defaults, which is `false`.
Then you have to provide a custom bean of type `KafkaBindingRebalanceListener`, which will be injected into all consumer bindings.
It is an interface that comes with a few default methods, but here is the method that we are interested in:
```
/**
* Invoked when partitions are initially assigned or after a rebalance. Applications
* might only want to perform seek operations on an initial assignment. While the
* 'initial' argument is true for each thread (when concurrency is greater than 1),
* implementations should keep track of exactly which partitions have been sought.
* There is a race in that a rebalance could occur during startup and so a topic/
* partition that has been sought on one thread may be re-assigned to another
* thread and you may not wish to re-seek it at that time.
* @param bindingName the name of the binding.
* @param consumer the consumer.
* @param partitions the partitions.
* @param initial true if this is the initial assignment on the current thread.
*/
default void onPartitionsAssigned(String bindingName, Consumer<?, ?> consumer,
Collection<TopicPartition> partitions, boolean initial) {
// do nothing
}
```
Let us look at the details.
In essence, this method will be invoked each time during the initial assignment for a topic partition or after a rebalance.
For better illustration, let us assume that our topic is `foo` and it has 4 partitions.
Initially, we are only starting a single consumer in the group and this consumer will consume from all partitions.
When the consumer starts for the first time, all 4 partitions are getting initially assigned.
However, we do not want to start the partitions to consume at the defaults (`earliest` since we define a group), rather for each partition, we want them to consume after seeking to arbitrary offsets.
Imagine that you have a business case to consume from certain offsets as below.
```
Partition start offset
0 1000
1 2000
2 2000
3 1000
```
This could be achieved by implementing the above method as below.
```
@Override
public void onPartitionsAssigned(String bindingName, Consumer<?, ?> consumer, Collection<TopicPartition> partitions, boolean initial) {
Map<TopicPartition, Long> topicPartitionOffset = new HashMap<>();
topicPartitionOffset.put(new TopicPartition("foo", 0), 1000L);
topicPartitionOffset.put(new TopicPartition("foo", 1), 2000L);
topicPartitionOffset.put(new TopicPartition("foo", 2), 2000L);
topicPartitionOffset.put(new TopicPartition("foo", 3), 1000L);
if (initial) {
partitions.forEach(tp -> {
if (topicPartitionOffset.containsKey(tp)) {
final Long offset = topicPartitionOffset.get(tp);
try {
consumer.seek(tp, offset);
}
catch (Exception e) {
// Handle exceptions carefully.
}
}
});
}
}
```
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?
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`.
When this new Spring Cloud Stream consumer calls this `onPartitionsAssigned` method, it will see that this is the initial assignment for partition `2` and `3` on this consumer.
Therefore, it will do the seek operation because of the conditional check on the `initial` argument.
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?
==== Problem Statement
Using Kafka binder, I want to manually acknowledge messages in my consumer.
How do I do that?
==== 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`.
See https://docs.spring.io/spring-kafka/docs/current/reference/html/#committing-offsets[here] for more details on that.
There are situations in which you want to disable this default commit behavior and rely on manual commits.
Following steps allow you to do that.
Set the property `spring.cloud.stream.kafka.bindings.<binding-name>.consumer.ackMode` to either `MANUAL` or `MANUAL_IMMEDIATE`.
When it is set like that, then there will be a header called `kafka_acknowledgment` (from `KafkaHeaders.ACKNOWLEDGMENT`) present in the message received by the consumer method.
For example, imagine this as your consumer method.
```
@Bean
public Consumer<Message<String>> myConsumer() {
return msg -> {
Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
if (acknowledgment != null) {
System.out.println("Acknowledgment provided");
acknowledgment.acknowledge();
}
};
}
```
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?
==== 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
Assume that following is your function signature.
```
@Bean
public Function<String, String> uppercase(){
...
}
```
By default, Spring Cloud Stream will create the bindings as below.
1. uppercase-in-0
2. uppercase-out-0
You can override these bindings to something by using the following properties.
```
spring.cloud.stream.function.bindings.uppercase-in-0=my-transformer-in
spring.cloud.stream.function.bindings.uppercase-out-0=my-transformer-out
```
After this, all binding properties must be made on the new names, `my-transformer-in` and `my-transformer-out`.
Here is another example with Kafka Streams and multiple inputs.
```
@Bean
public BiFunction<KStream<String, Order>, KTable<String, Account>, KStream<String, EnrichedOrder>> processOrder() {
...
}
```
By default, Spring Cloud Stream will create three different binding names for this function.
1. processOrder-in-0
2. processOrder-in-1
3. processOrder-out-0
You have to use these binding names each time you want to set some configuration on these bindings.
You don't like that, and you want to use more domain-friendly and readable binding names, for example, something like.
1. orders
2. accounts
3. enrichedOrders
You can easily do that by simply setting these three properties
1. spring.cloud.stream.function.bindings.processOrder-in-0=orders
2. spring.cloud.stream.function.bindings.processOrder-in-1=accounts
3. spring.cloud.stream.function.bindings.processOrder-out-0=enrichedOrders
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?
==== 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
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.
Following is a basic blueprint for doing this, but you may want to adapt it to your paricular use case.
Here is sample producer method (aka `Supplier`).
```
@Bean
public Supplier<Message<String>> supplier() {
return () -> MessageBuilder.withPayload("foo").setHeader(KafkaHeaders.MESSAGE_KEY, "my-foo").build();
}
```
This is a trivial function that sends a message with a `String` payload, but also with a key.
Note that we set the key as a message header using `KafkaHeaders.MESSAGE_KEY`.
If you want to change the key from the default `kafka_messageKey`, then in the configuration, we need to specify this property:
```
spring.cloud.stream.kafka.bindings.supplier-out-0.producer.messageKeyExpression=headers['my-special-key']
```
Please note that we use the binding name `supplier-out-0` since that is our function name, please update accordingly.
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?
==== 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
This is really easy to do.
All you have to do is to provide the following property to enable native serialization.
```
spring.cloud.stream.kafka.bindings.<binding-name>.producer.useNativeEncoding: true
```
Then, you need to also set the serializers.
There are a couple of ways to do this.
```
spring.cloud.stream.kafka.bindings.<binding-name>.producer.configuration.key.serializer: org.apache.kafka.common.serialization.StringSerializer
spring.cloud.stream.kafka.bindings.<binding-name>.producer.configuration.value.serializer: org.apache.kafka.common.serialization.StringSerializer
```
or using the binder configuration.
```
spring.cloud.stream.kafka.binder.configuration.key.serializer: org.apache.kafka.common.serialization.StringSerializer
spring.cloud.stream.kafka.binder.configuration.value.serializer: org.apache.kafka.common.serialization.StringSerializer
```
When using the binder way, it is applied against all the bindings whereas setting them at the bindings are per binding.
On the deserializing side, you just need to provide the deserializers as configuration.
For example,
```
spring.cloud.stream.kafka.bindings.<binding-name>.consumer.configuration.key.deserializer: org.apache.kafka.common.serialization.StringDeserializer
spring.cloud.stream.kafka.bindings.<binding-name>.producer.configuration.value.deserializer: org.apache.kafka.common.serialization.StringDeserializer
```
You can also set them at the binder level.
There is an optional property that you can set to force native decoding.
```
spring.cloud.stream.kafka.bindings.<binding-name>.consumer.useNativeDecoding: true
```
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
==== 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
Before we look at the solution, let us look at the following scenario.
```
@Bean
public BiConsumer<KStream<Object, Object>, KTable<Object, Object>> myBiConsumer{
(s, t) -> s.join(t, ...)
...
}
```
We have a `BiConsumer` bean that requires two input bindings.
In this case, the first binding is for a `KStream` and the second one is for a `KTable`.
When running this application for the first time, by default, both bindings start from the `earliest` offset.
What about I want to start from the `latest` offset due to some requirements?
You can do this by enabling the following properties.
```
spring.cloud.stream.kafka.streams.bindings.myBiConsumer-in-0.consumer.startOffset: latest
spring.cloud.stream.kafka.streams.bindings.myBiConsumer-in-1.consumer.startOffset: latest
```
If you want only one binding to start from the `latest` offset and the other to consumer from the default `earliest`, then leave the latter binding out from the configuration.
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
==== Problem Statement
I have a Kafka producer application and I want to keep track of all my successful sendings.
==== Solution
Let us assume that we have this following supplier in the application.
```
@Bean
public Supplier<Message<String>> supplier() {
return () -> MessageBuilder.withPayload("foo").setHeader(KafkaHeaders.MESSAGE_KEY, "my-foo").build();
}
```
Then, we need to define a new `MessageChannel` bean to capture all the successful send information.
```
@Bean
public MessageChannel fooRecordChannel() {
return new DirectChannel();
}
```
Next, define this property in the application configuration to provide the bean name for the `recordMetadataChannel`.
```
spring.cloud.stream.kafka.bindings.supplier-out-0.producer.recordMetadataChannel: fooRecordChannel
```
At this point, successful sent information will be sent to the `fooRecordChannel`.
You can write an `IntegrationFlow` as below to see the information.
```
@Bean
public IntegrationFlow integrationFlow() {
return f -> f.channel("fooRecordChannel")
.handle((payload, messageHeaders) -> payload);
}
```
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
==== Problem Statement
I have a Kafka producer application that sets some headers, but they are missing in the consumer application. Why is that?
==== Solution
Under normal circumstances, this should be fine.
Imagine, you have the following producer.
```
@Bean
public Supplier<Message<String>> supply() {
return () -> MessageBuilder.withPayload("foo").setHeader("foo", "bar").build();
}
```
On the consumer side, you should still see the header "foo", and the following should not give you any issues.
```
@Bean
public Consumer<Message<String>> consume() {
return s -> {
final String foo = (String)s.getHeaders().get("foo");
System.out.println(foo);
};
}
```
If you provide a https://docs.spring.io/spring-cloud-stream-binder-kafka/docs/3.1.3/reference/html/spring-cloud-stream-binder-kafka.html#_kafka_binder_properties[custom header mapper] in the application, then this won't work.
Let's say you have an empty `KafkaHeaderMapper` in the application.
```
@Bean
public KafkaHeaderMapper kafkaBinderHeaderMapper() {
return new KafkaHeaderMapper() {
@Override
public void fromHeaders(MessageHeaders headers, Headers target) {
}
@Override
public void toHeaders(Headers source, Map<String, Object> target) {
}
};
}
```
If that is your implementation, then you will miss the `foo` header on the consumer.
Chances are that, you may have some logic inside those `KafkaHeaderMapper` methods.
You need the following to populate the `foo` header.
```
@Bean
public KafkaHeaderMapper kafkaBinderHeaderMapper() {
return new KafkaHeaderMapper() {
@Override
public void fromHeaders(MessageHeaders headers, Headers target) {
final String foo = (String) headers.get("foo");
target.add("foo", foo.getBytes());
}
@Override
public void toHeaders(Headers source, Map<String, Object> target) {
final Header foo = source.lastHeader("foo");
target.put("foo", new String(foo.value()));
}
}
```
That will properly populate the `foo` header from the producer to consumer.
==== 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.
However, if you produce with a variant of the framework reserved `id` header - such as `ID`, `Id`, `iD` etc. then you will run into issues with the internals of the framework.
See this https://stackoverflow.com/questions/68412600/change-the-behaviour-in-spring-cloud-stream-make-header-matcher-case-sensitive[StackOverflow thread] fore more context on this use case.
In that case, you must use a custom `KafkaHeaderMapper` to map the case-sensitive id header.
For example, let's say you have the following producer.
```
@Bean
public Supplier<Message<String>> supply() {
return () -> MessageBuilder.withPayload("foo").setHeader("Id", "my-id").build();
}
```
The header `Id` above will be gone from the consuming side as it clashes with the framework `id` header.
You can provide a custom `KafkaHeaderMapper` to solve this issue.
```
@Bean
public KafkaHeaderMapper kafkaBinderHeaderMapper1() {
return new KafkaHeaderMapper() {
@Override
public void fromHeaders(MessageHeaders headers, Headers target) {
final String myId = (String) headers.get("Id");
target.add("Id", myId.getBytes());
}
@Override
public void toHeaders(Headers source, Map<String, Object> target) {
final Header Id = source.lastHeader("Id");
target.put("Id", new String(Id.value()));
}
};
}
```
By doing this, both `id` and `Id` headers will be available from the producer to the consumer side.
=== Producing to multiple topics in transaction
==== 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
Use transactional support in Kafka binder for transactions and then provide an `AfterRollbackProcessor`.
In order to produce to multiple topics, use `StreamBridge` API.
Below are the code snippets for this:
```
@Autowired
StreamBridge bridge;
@Bean
Consumer<String> input() {
return str -> {
System.out.println(str);
this.bridge.send("left", str.toUpperCase());
this.bridge.send("right", str.toLowerCase());
if (str.equals("Fail")) {
throw new RuntimeException("test");
}
};
}
@Bean
ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> customizer(BinderFactory binders) {
return (container, dest, group) -> {
ProducerFactory<byte[], byte[]> pf = ((KafkaMessageChannelBinder) binders.getBinder(null,
MessageChannel.class)).getTransactionalProducerFactory();
KafkaTemplate<byte[], byte[]> template = new KafkaTemplate<>(pf);
DefaultAfterRollbackProcessor rollbackProcessor = rollbackProcessor(template);
container.setAfterRollbackProcessor(rollbackProcessor);
};
}
DefaultAfterRollbackProcessor rollbackProcessor(KafkaTemplate<byte[], byte[]> template) {
return new DefaultAfterRollbackProcessor<>(
new DeadLetterPublishingRecoverer(template), new FixedBackOff(2000L, 2L), template, true);
}
```
==== Required Configuration
```
spring.cloud.stream.kafka.binder.transaction.transaction-id-prefix: tx-
spring.cloud.stream.kafka.binder.required-acks=all
spring.cloud.stream.bindings.input-in-0.group=foo
spring.cloud.stream.bindings.input-in-0.destination=input
spring.cloud.stream.bindings.left.destination=left
spring.cloud.stream.bindings.right.destination=right
spring.cloud.stream.kafka.bindings.input-in-0.consumer.maxAttempts=1
```
in order to test, you can use the following:
```
@Bean
public ApplicationRunner runner(KafkaTemplate<byte[], byte[]> template) {
return args -> {
System.in.read();
template.send("input", "Fail".getBytes());
template.send("input", "Good".getBytes());
};
}
```
Some important notes:
Please ensure that you don't have any DLQ settings on the application configuration as we manually configure DLT (By default it will be published to a topic named `input.DLT` based on the initial consumer function).
Also, reset the `maxAttempts` on consumer binding to `1` in order to avoid retries by the binder.
It will be max tried a total of 3 in the example above (initial try + the 2 attempts in the `FixedBackoff`).
See the https://stackoverflow.com/questions/68928091/dlq-bounded-retry-and-eos-when-producing-to-multiple-topics-using-spring-cloud[StackOverflow thread] for more details on how to test this code.
If you are using Spring Cloud Stream to test it by adding more consumer functions, make sure to set the `isolation-level` on the consumer binding to `read-committed`.
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
==== Problem Statement
How can I run multiple instances of the pollable consumers and generate unique `client.id` for each instance?
==== Solution
Assuming that I have the following definition:
```
spring.cloud.stream.pollable-source: foo
spring.cloud.stream.bindings.foo-in-0.group: my-group
```
When running the application, the Kafka consumer generates a client.id (something like `consumer-my-group-1`).
For each instance of the application that is running, this `client.id` will be the same, causing unexpected issues.
In order to fix this, you can add the following property on each instance of the application:
```
spring.cloud.stream.kafka.bindings.foo-in-0.consumer.configuration.client.id=${client.id}
```
See this https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1139[GitHub issue] for more details.

49
docs/modules/ROOT/pages/kafka/spring-cloud-stream-binder-kafka.adoc Normal file
View File

@@ -0,0 +1,49 @@
:github-tag: master
:github-repo: spring-cloud/spring-cloud-stream-binder-kafka
:github-raw: https://raw.githubusercontent.com/{github-repo}/{github-tag}
:github-code: https://github.com/{github-repo}/tree/{github-tag}
:toc: left
:toclevels: 8
:nofooter:
:sectlinks: true
[[spring-cloud-stream-binder-kafka-reference]]
= Spring Cloud Stream Kafka Binder Reference Guide
Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinathan, Gunnar Hillert, Mark Pollack, Patrick Peralta, Glenn Renfro, Thomas Risberg, Dave Syer, David Turanski, Janne Valkealahti, Benjamin Klein, Henryk Konsek, Gary Russell, Arnaud Jardiné, Soby Chacko
:doctype: book
:toc:
:toclevels: 4
:source-highlighter: prettify
:numbered:
:icons: font
:hide-uri-scheme:
:spring-cloud-stream-binder-kafka-repo: snapshot
:github-tag: master
:spring-cloud-stream-binder-kafka-docs-version: current
:spring-cloud-stream-binder-kafka-docs: https://docs.spring.io/spring-cloud-stream-binder-kafka/docs/{spring-cloud-stream-binder-kafka-docs-version}/reference
:spring-cloud-stream-binder-kafka-docs-current: https://docs.spring.io/spring-cloud-stream-binder-kafka/docs/current-SNAPSHOT/reference/html/
:github-repo: spring-cloud/spring-cloud-stream-binder-kafka
:github-raw: https://raw.github.com/{github-repo}/{github-tag}
:github-code: https://github.com/{github-repo}/tree/{github-tag}
:github-wiki: https://github.com/{github-repo}/wiki
:github-master-code: https://github.com/{github-repo}/tree/master
:sc-ext: java
// ======================================================================================
*{project-version}*
= Reference Guide
include::kafka/kafka_overview.adoc[]
include::kafka/kafka_dlq.adoc[]
include::kafka/kafka_partitions.adoc[]
include::kafka/kafka-reactive.adoc[]
include::kafka/kafka-streams.adoc[]
include::kafka/kafka_tips.adoc[]

171
docs/modules/ROOT/pages/preface.adoc Normal file
View File

@@ -0,0 +1,171 @@
=== 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.
Fast forward to the cloud-era, where microservices have become prominent in the enterprise setting. https://projects.spring.io/spring-boot/[Spring Boot] transformed the way how developers built Applications. With Spring's programming model and the runtime responsibilities handled by Spring Boot, it became seamless to develop stand-alone, production-grade Spring-based microservices.
To extend this to Data Integration workloads, Spring Integration and Spring Boot were put together into a new project. Spring Cloud Stream was born.
[%hardbreaks]
With Spring Cloud Stream, developers can:
- Build, test and deploy data-centric applications in isolation.
- Apply modern microservices architecture patterns, including composition through messaging.
- Decouple application responsibilities with event-centric thinking. An event can represent something that has happened in time, to which the downstream consumer applications can react without knowing where it originated or the producer's identity.
- Port the business logic onto message brokers (such as RabbitMQ, Apache Kafka, Amazon Kinesis).
- 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
You can try Spring Cloud Stream in less than 5 min even before you jump into any details by following this three-step guide.
We show you how to create a Spring Cloud Stream application that receives messages coming from the messaging middleware of your choice (more on this later) and logs received messages to the console.
We call it `LoggingConsumer`.
While not very practical, it provides a good introduction to some of the main concepts
and abstractions, making it easier to digest the rest of this user guide.
The three steps are as follows:
. <<spring-cloud-stream-preface-creating-sample-application>>
. <<spring-cloud-stream-preface-importing-project>>
. <<spring-cloud-stream-preface-adding-message-handler>>
[[spring-cloud-stream-preface-creating-sample-application]]
==== 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`.
When the "`Cloud Stream`" option should appears, select it.
. Start typing either 'kafka' or 'rabbit'.
. Select "`Kafka`" or "`RabbitMQ`".
+
Basically, you choose the messaging middleware to which your application binds.
We recommend using the one you have already installed or feel more comfortable with installing and running.
Also, as you can see from the Initilaizer screen, there are a few other options you can choose.
For example, you can choose Gradle as your build tool instead of Maven (the default).
. In the *Artifact* field, type 'logging-consumer'.
+
The value of the *Artifact* field becomes the application name.
If you chose RabbitMQ for the middleware, your Spring Initializr should now be as follows:
[%hardbreaks]
[%hardbreaks]
[%hardbreaks]
image::spring-initializr.png[align="center"]
[%hardbreaks]
[%hardbreaks]
. Click the *Generate Project* button.
+
Doing so downloads the zipped version of the generated project to your hard drive.
. Unzip the file into the folder you want to use as your project directory.
TIP: We encourage you to explore the many possibilities available in the Spring Initializr.
It lets you create many different kinds of Spring applications.
[[spring-cloud-stream-preface-importing-project]]
==== 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.
For example, depending on how the project was generated (Maven or Gradle), you may need to follow specific import procedure (for example, in Eclipse or STS, you need to use File -> Import -> Maven -> Existing Maven Project).
Once imported, the project must have no errors of any kind. Also, `src/main/java` should contain `com.example.loggingconsumer.LoggingConsumerApplication`.
Technically, at this point, you can run the application's main class.
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
Modify the `com.example.loggingconsumer.LoggingConsumerApplication` class to look as follows:
[source, java]
----
@SpringBootApplication
public class LoggingConsumerApplication {
public static void main(String[] args) {
SpringApplication.run(LoggingConsumerApplication.class, args);
}
@Bean
public Consumer<Person> log() {
return person -> {
System.out.println("Received: " + person);
};
}
public static class Person {
private String name;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String toString() {
return this.name;
}
}
}
----
As you can see from the preceding listing:
* We are using functional programming model (see <<Spring Cloud Function support>>) to define a single message handler as `Consumer`.
* We are relying on framework conventions to bind such handler to the input destination binding exposed by the binder.
Doing so also lets you see one of the core features of the framework: It tries to automatically convert incoming message payloads to type `Person`.
You now have a fully functional Spring Cloud Stream application that does listens for messages.
From here, for simplicity, we assume you selected RabbitMQ in <<spring-cloud-stream-preface-creating-sample-application,step one>>.
Assuming you have RabbitMQ installed and running, you can start the application by running its `main` method in your IDE.
You should see following output:
[source]
----
--- [ main] c.s.b.r.p.RabbitExchangeQueueProvisioner : declaring queue for inbound: input.anonymous.CbMIwdkJSBO1ZoPDOtHtCg, bound to: input
--- [ main] o.s.a.r.c.CachingConnectionFactory : Attempting to connect to: [localhost:5672]
--- [ main] o.s.a.r.c.CachingConnectionFactory : Created new connection: rabbitConnectionFactory#2a3a299:0/SimpleConnection@66c83fc8. . .
. . .
--- [ main] o.s.i.a.i.AmqpInboundChannelAdapter : started inbound.input.anonymous.CbMIwdkJSBO1ZoPDOtHtCg
. . .
--- [ main] c.e.l.LoggingConsumerApplication : Started LoggingConsumerApplication in 2.531 seconds (JVM running for 2.897)
----
Go to the RabbitMQ management console or any other RabbitMQ client and send a message to `input.anonymous.CbMIwdkJSBO1ZoPDOtHtCg`.
The `anonymous.CbMIwdkJSBO1ZoPDOtHtCg` part represents the group name and is generated, so it is bound to be different in your environment.
For something more predictable, you can use an explicit group name by setting `spring.cloud.stream.bindings.input.group=hello` (or whatever name you like).
The contents of the message should be a JSON representation of the `Person` class, as follows:
{"name":"Sam Spade"}
Then, in your console, you should see:
`Received: Sam Spade`
You can also build and package your application into a boot jar (by using `./mvnw clean install`) and run the built JAR by using the `java -jar` command.
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
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.
SpEL gives you access to the current Message as well as the Application Context you are running in.
However it is important to understand what type of data SpEL can see especially in the context of the incoming Message.
From the broker, the message arrives in a form of a byte[]. It is then transformed to a `Message<byte[]>` by the binders where as you can see the payload of the message maintains its raw form. The headers of the message are `<String, Object>`, where values are typically another primitive or a collection/array of primitives, hence Object.
That is because binder does not know the required input type as it has no access to the user code (function). So effectively binder delivered an envelope with the payload and some readable meta-data in the form of message headers, just like the letter delivered by mail.
This means that while accessing payload of the message is possible you will only have access to it as raw data (i.e., byte[]). And while it may be very common for developers to ask for ability to have SpEL access to fields of a payload object as concrete type (e.g., Foo, Bar etc), you can see how difficult or even impossible would it be to achieve.
Here is one example to demonstrate the problem; Imagine you have a routing expression to route to different functions based on payload type. This requirement would imply payload conversion from byte[] to a specific type and then applying the SpEL. However, in order to perform such conversion we would need to know the actual type to pass to converter and that comes from function's signature which we don’t know which one. A better approach to solve this requirement would be to pass the type information as message headers (e.g., `application/json;type=foo.bar.Baz` ). You’ll get a clear readable String value that could be accessed and evaluated in a year and easy to read SpEL expression.
Additionally it is considered very bad practice to use payload for routing decisions, since the payload is considered to be privileged data - data only to be read by its final recipient. Again, using the mail delivery analogy you would not want the mailman to open your envelope and read the contents of the letter to make some delivery decisions. The same concept applies here, especially when it is relatively easy to include such information when generating a Message. It enforces certain level of discipline related to the design of data to be transmitted over the network and which pieces of such data can be considered as public and which are privileged.

258
docs/modules/ROOT/pages/pulsar/pulsar_binder.adoc Normal file
View File

@@ -0,0 +1,258 @@
[[pulsar-binder]]
= Spring Cloud Stream Binder for Apache Pulsar
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
We need to include the following dependency on your application to use Apache Pulsar binder for Spring Cloud Stream.
[source,xml,indent=0,subs="verbatim,attributes",role="primary"]
.Maven
----
<dependencies>
<dependency>
<groupId>org.springframework.pulsar</groupId>
<artifactId>spring-pulsar-spring-cloud-stream-binder</artifactId>
<version>{spring-pulsar-version}</version>
</dependency>
</dependencies>
----
[source,groovy,indent=0,subs="verbatim,attributes",role="secondary"]
.Gradle
----
dependencies {
implementation 'org.springframework.pulsar:spring-pulsar-spring-cloud-stream-binder:{spring-pulsar-version}'
}
----
== 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.
The binder takes care of all those details for the application developer.
Spring Cloud Stream brings a powerful programming model based on {spring-cloud-function}[Spring Cloud Function] that allows the app developer to write complex event-driven applications using a functional style.
Applications can start from a middleware-neutral manner and then map Pulsar topics as destinations in Spring Cloud Stream through Spring Boot configuration properties.
Spring Cloud Stream is built on top of Spring Boot, and when writing an event-driven microservice using Spring Cloud Stream, you are essentially writing a Boot application.
Here is a straightforward Spring Cloud Stream application.
====
[source, java]
----
@SpringBootApplication
public class SpringPulsarBinderSampleApp {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
public static void main(String[] args) {
SpringApplication.run(SpringPulsarBinderSampleApp.class, args);
}
@Bean
public Supplier<Time> timeSupplier() {
return () -> new Time(String.valueOf(System.currentTimeMillis()));
}
@Bean
public Function<Time, EnhancedTime> timeProcessor() {
return (time) -> {
EnhancedTime enhancedTime = new EnhancedTime(time, "5150");
this.logger.info("PROCESSOR: {} --> {}", time, enhancedTime);
return enhancedTime;
};
}
@Bean
public Consumer<EnhancedTime> timeLogger() {
return (time) -> this.logger.info("SINK: {}", time);
}
record Time(String time) {
}
record EnhancedTime(Time time, String extra) {
}
}
----
====
The above sample application, a full-blown Spring Boot application, deserves a few explanations. However, on a first pass, you can see that this is just plain Java and a few Spring and Spring Boot annotations.
We have three `Bean` methods here - a `java.util.function.Supplier`, a `java.util.function.Function`, and finally, a `java.util.function.Consumer`.
The supplier produces the current time in milliseconds, the function takes this time and then enhances it by adding some random data, and then the consumer logs the enhanced time.
We omitted all the imports for brevity, but nothing Spring Cloud Stream specific in the entire application.
How does it become a Spring Cloud Stream application that interacts with Apache Pulsar?
You must include the above dependency for the binder in the application.
Once that dependency is added, you must provide the following configuration properties.
[source,yaml,indent=0,subs="verbatim"]
----
spring:
cloud:
function:
definition: timeSupplier;timeProcessor;timeLogger;
stream:
bindings:
timeProcessor-in-0:
destination: timeSupplier-out-0
timeProcessor-out-0:
destination: timeProcessor-out-0
timeLogger-in-0:
destination: timeProcessor-out-0
----
With this, the above Spring Boot application has become an end-to-end event-driven application based on Spring Cloud Stream.
Because we have the Pulsar binder on the classpath, the application interacts with Apache Pulsar.
If there is only one function in the application, then we don't need to tell Spring Cloud Stream to activate the function for execution since it does that by default.
If there is more than one such function in the application, as in our example, we need to instruct Spring Cloud Stream which functions we would like to activate.
In our case, we need all of them to be activated, and we do that through the `spring.cloud.function.definition` property.
The bean name becomes part of the Spring Cloud Stream binding name by default.
A binding is a fundamentally abstract concept in Spring Cloud Stream, using which the framework communicates with the middleware destination.
Almost everything that Spring Cloud Stream does occurs over a concrete binding.
A supplier has only an output binding; functions have input and output bindings, and consumers have only input binding.
Let's take as an example our supplier bean - `timeSupplier.`
The default binding name for this supplier will be `timeSupplier-out-0`.
Similarly, the default binding names for the `timeProcessor` function will be `timeProcessor-in-0` on the inbound and `timeProcessor-out-0` on the outbound.
Please refer to the Spring Cloud Stream reference docs for details on how you can change the default binding names.
In most situations, using the default binding names is enough.
We set the destination on the binding names, as shown above.
If a destination is not provided, the binding name becomes the value for the destination as in the case of `timeSupplier-out-0`.
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
In the above sample application, we provided no schema information for message conversion.
That is because, by default, Spring Cloud Stream uses its message conversion mechanism using the messaging support established in Spring Framework through the Spring Messaging project.
Unless specified, Spring Cloud Stream uses `application/json` as the `content-type` for message conversion on both inbound and outbound bindings.
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
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.
Suppose the application chooses to go this route. In that case, Spring Cloud Stream steers clear of using any Spring-provided message conversion facility and passes around the data it receives or produces.
This feature in Spring Cloud Stream is known as native encoding on the producer side and native decoding on the consumer side. This means that the encoding and decoding natively occur on the target middleware, in our case, on Apache Pulsar.
For the above application, we can use the following configuration to bypass the framework conversion and uses native encoding and decoding.
[source,yaml,indent=0,subs="verbatim"]
----
spring:
cloud:
stream:
bindings:
timeSupplier-out-0:
producer:
use-native-encoding: true
timeProcessor-in-0:
destination: timeSupplier-out-0
consumer:
use-native-decoding: true
timeProcessor-out-0:
destination: timeProcessor-out-0
producer:
use-native-encoding: true
timeLogger-in-0:
destination: timeProcessor-out-0
consumer:
use-native-decoding: true
pulsar:
bindings:
timeSupplier-out-0:
producer:
schema-type: JSON
message-type: org.springframework.pulsar.sample.binder.SpringPulsarBinderSampleApp.Time
timeProcessor-in-0:
consumer:
schema-type: JSON
message-type: org.springframework.pulsar.sample.binder.SpringPulsarBinderSampleApp.Time
timeProcessor-out-0:
producer:
schema-type: AVRO
message-type: org.springframework.pulsar.sample.binder.SpringPulsarBinderSampleApp.EnhancedTime
timeLogger-in-0:
consumer:
schema-type: AVRO
message-type: org.springframework.pulsar.sample.binder.SpringPulsarBinderSampleApp.EnhancedTime
----
The property to enable native encoding on the producer side is a binding level property from the core Spring Cloud Stream.
You set it on the producer binding - `spring.cloud.stream.bindings.<binding-name>.producer.use-native-encoding` and set this to `true.`
Similarly, use - `spring.cloud.stream.bindings.<binding-name>.consumer.user-native-decoding` for consumer bindings and set it to `true.`
If we decide to use native encoding and decoding, in the case of Pulsar, we need to set the corresponding schema and the underlying message type information.
This information is provided as extended binding properties.
As you can see above in the configuration, the properties are - `spring.cloud.stream.pulsar.bindings.<binding-name>.producer|consumer.schema-type` for schema information and `spring.cloud.stream.pulsar.bindings.<binding-name>.producer|consumer.message-type` for the actual target type.
If you have both keys and values on the message, you can use `message-key-type` and `message-value-type` to specify their target types.
TIP: Any configured custom schema mappings will be consulted when the `schema-type` property is omitted.
=== 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
The Pulsar binder is configured with a default header mapper that can be overridden by providing your own `PulsarHeaderMapper` bean.
In the following example, a JSON header mapper is configured that:
- maps all inbound headers (except those with keys "`top`" or "`secret`")
- maps outbound headers (except those with keys "`id`", "`timestamp`", or "`userId`")
- only trusts objects in the "`com.acme`" package for outbound deserialization
- de/serializes any "`com.acme.Money`" header values w/ simple `toString()` encoding
[source,java,indent=0]
----
@Bean
public PulsarHeaderMapper customPulsarHeaderMapper() {
return JsonPulsarHeaderMapper.builder()
.inboundPatterns("!top", "!secret", "*")
.outboundPatterns("!id", "!timestamp", "!userId", "*")
.trustedPackages("com.acme")
.toStringClasses("com.acme.Money")
.build();
}
----
== Using Pulsar Properties in the Binder
The binder uses basic components from Spring for Apache Pulsar framework to build its producer and consumer bindings.
Since binder-based applications are Spring Boot applications, binder, by default, uses the Spring Boot autoconfiguration for Spring for Apache Pulsar.
Therefore, all Pulsar Spring Boot properties available at the core framework level are also available through the binder.
For example, you can use properties with the prefix `spring.pulsar.producer...`, `spring.pulsar.consumer...` etc.
In addition, you can also set these Pulsar properties at the binder level.
For instance, this will also work - `spring.cloud.stream.pulsar.binder.producer...` or `spring.cloud.stream.pulsar.binder.consumer...`.
Either of the above approaches is fine, but when using properties like these, it is applied to the whole application.
If you have multiple functions in the application, they all get the same properties.
You can also set these Pulsar properties at the extended binding properties level to address this.
Extended binding properties are applied at the binding itself.
For instance, if you have an input and output binding, and both require a separate set of Pulsar properties, you must set them on the extended binding.
The pattern for producer binding is `spring.cloud.stream.pulsar.bindings.<output-binding-name>.producer...`.
Similarly, for consumer binding, the pattern is `spring.cloud.stream.pulsar.bindings.<input-binding-name>.consumer...`.
This way, you can have a separate set of Pulsar properties applied for different bindings in the same application.
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
Spring Cloud Stream binder for Apache Pulsar comes with an out-of-the-box provisioner for Pulsar topics.
When running an application, if the necessary topics are absent, Pulsar will create the topics for you.
However, this is a basic non-partitioned topic, and if you want advanced features like creating a partitioned topic, you can rely on the topic provisioner in the binder.
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
When creating the topic, you can set the partition count in two ways.
First, you can set it at the binder level using the property `spring.cloud.stream.pulsar.binder.partition-count`.
As we saw above, doing this way will make all the topics created by the application inherit this property.
Suppose you want granular control at the binding level for setting partitions.
In that case, you can set the `partition-count` property per binding using the format `spring.cloud.stream.pulsar.bindings.<binding-name>.producer|consumer.partition-count`.
This way, various topics created by different functions in the same application will have different partitions based on the application requirements.

24
docs/modules/ROOT/pages/pulsar/spring-cloud-stream-binder-pulsar.adoc Normal file
View File

@@ -0,0 +1,24 @@
:toc: left
:toclevels: 8
:nofooter:
:sectlinks: true
[[spring-cloud-stream-binder-pulsar-reference]]
= Spring Cloud Stream Pulsar Binder Reference Guide
Soby Chacko; Chris Bono; Alexander Preuß; Jay Bryant; Christophe Bornet
:doctype: book
:toc:
:toclevels: 4
:source-highlighter: prettify
:numbered:
:icons: font
:hide-uri-scheme:
:sc-ext: java
// ======================================================================================
*{project-version}*
= Reference Guide
include::pulsar/pulsar_binder.adoc[]

243
docs/modules/ROOT/pages/rabbit/rabbit_dlq.adoc Normal file
View File

@@ -0,0 +1,243 @@
[[rabbit-dlq-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.
However, if the problem is a permanent issue, that could cause an infinite loop.
The following Spring Boot application shows an example of how to route those messages back to the original queue but moves them to a third "`parking lot`" queue after three attempts.
The second example uses the https://www.rabbitmq.com/blog/2015/04/16/scheduling-messages-with-rabbitmq/[RabbitMQ Delayed Message Exchange] to introduce a delay to the re-queued message.
In this example, the delay increases for each attempt.
These examples use a `@RabbitListener` to receive messages from the DLQ.
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
The first two examples are for when the destination is *not* partitioned:
[source, java]
----
@SpringBootApplication
public class ReRouteDlqApplication {
private static final String ORIGINAL_QUEUE = "so8400in.so8400";
private static final String DLQ = ORIGINAL_QUEUE + ".dlq";
private static final String PARKING_LOT = ORIGINAL_QUEUE + ".parkingLot";
private static final String X_RETRIES_HEADER = "x-retries";
public static void main(String[] args) throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(ReRouteDlqApplication.class, args);
System.out.println("Press enter to exit");
System.in.read();
context.close();
}
@Autowired
private RabbitTemplate rabbitTemplate;
@RabbitListener(queues = DLQ)
public void rePublish(Message failedMessage) {
Integer retriesHeader = (Integer) failedMessage.getMessageProperties().getHeaders().get(X_RETRIES_HEADER);
if (retriesHeader == null) {
retriesHeader = Integer.valueOf(0);
}
if (retriesHeader < 3) {
failedMessage.getMessageProperties().getHeaders().put(X_RETRIES_HEADER, retriesHeader + 1);
this.rabbitTemplate.send(ORIGINAL_QUEUE, failedMessage);
}
else {
this.rabbitTemplate.send(PARKING_LOT, failedMessage);
}
}
@Bean
public Queue parkingLot() {
return new Queue(PARKING_LOT);
}
}
----
[source, java]
----
@SpringBootApplication
public class ReRouteDlqApplication {
private static final String ORIGINAL_QUEUE = "so8400in.so8400";
private static final String DLQ = ORIGINAL_QUEUE + ".dlq";
private static final String PARKING_LOT = ORIGINAL_QUEUE + ".parkingLot";
private static final String X_RETRIES_HEADER = "x-retries";
private static final String DELAY_EXCHANGE = "dlqReRouter";
public static void main(String[] args) throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(ReRouteDlqApplication.class, args);
System.out.println("Press enter to exit");
System.in.read();
context.close();
}
@Autowired
private RabbitTemplate rabbitTemplate;
@RabbitListener(queues = DLQ)
public void rePublish(Message failedMessage) {
Map<String, Object> headers = failedMessage.getMessageProperties().getHeaders();
Integer retriesHeader = (Integer) headers.get(X_RETRIES_HEADER);
if (retriesHeader == null) {
retriesHeader = Integer.valueOf(0);
}
if (retriesHeader < 3) {
headers.put(X_RETRIES_HEADER, retriesHeader + 1);
headers.put("x-delay", 5000 * retriesHeader);
this.rabbitTemplate.send(DELAY_EXCHANGE, ORIGINAL_QUEUE, failedMessage);
}
else {
this.rabbitTemplate.send(PARKING_LOT, failedMessage);
}
}
@Bean
public DirectExchange delayExchange() {
DirectExchange exchange = new DirectExchange(DELAY_EXCHANGE);
exchange.setDelayed(true);
return exchange;
}
@Bean
public Binding bindOriginalToDelay() {
return BindingBuilder.bind(new Queue(ORIGINAL_QUEUE)).to(delayExchange()).with(ORIGINAL_QUEUE);
}
@Bean
public Queue parkingLot() {
return new Queue(PARKING_LOT);
}
}
----
=== Partitioned Destinations
With partitioned destinations, there is one DLQ for all partitions. We determine the original queue from the headers.
==== `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:
[source, java]
----
@SpringBootApplication
public class ReRouteDlqApplication {
private static final String ORIGINAL_QUEUE = "so8400in.so8400";
private static final String DLQ = ORIGINAL_QUEUE + ".dlq";
private static final String PARKING_LOT = ORIGINAL_QUEUE + ".parkingLot";
private static final String X_DEATH_HEADER = "x-death";
private static final String X_RETRIES_HEADER = "x-retries";
public static void main(String[] args) throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(ReRouteDlqApplication.class, args);
System.out.println("Press enter to exit");
System.in.read();
context.close();
}
@Autowired
private RabbitTemplate rabbitTemplate;
@SuppressWarnings("unchecked")
@RabbitListener(queues = DLQ)
public void rePublish(Message failedMessage) {
Map<String, Object> headers = failedMessage.getMessageProperties().getHeaders();
Integer retriesHeader = (Integer) headers.get(X_RETRIES_HEADER);
if (retriesHeader == null) {
retriesHeader = Integer.valueOf(0);
}
if (retriesHeader < 3) {
headers.put(X_RETRIES_HEADER, retriesHeader + 1);
List<Map<String, ?>> xDeath = (List<Map<String, ?>>) headers.get(X_DEATH_HEADER);
String exchange = (String) xDeath.get(0).get("exchange");
List<String> routingKeys = (List<String>) xDeath.get(0).get("routing-keys");
this.rabbitTemplate.send(exchange, routingKeys.get(0), failedMessage);
}
else {
this.rabbitTemplate.send(PARKING_LOT, failedMessage);
}
}
@Bean
public Queue parkingLot() {
return new Queue(PARKING_LOT);
}
}
----
==== `republishToDlq=true`
When `republishToDlq` is `true`, the republishing recoverer adds the original exchange and routing key to headers, as shown in the following example:
[source, java]
----
@SpringBootApplication
public class ReRouteDlqApplication {
private static final String ORIGINAL_QUEUE = "so8400in.so8400";
private static final String DLQ = ORIGINAL_QUEUE + ".dlq";
private static final String PARKING_LOT = ORIGINAL_QUEUE + ".parkingLot";
private static final String X_RETRIES_HEADER = "x-retries";
private static final String X_ORIGINAL_EXCHANGE_HEADER = RepublishMessageRecoverer.X_ORIGINAL_EXCHANGE;
private static final String X_ORIGINAL_ROUTING_KEY_HEADER = RepublishMessageRecoverer.X_ORIGINAL_ROUTING_KEY;
public static void main(String[] args) throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(ReRouteDlqApplication.class, args);
System.out.println("Press enter to exit");
System.in.read();
context.close();
}
@Autowired
private RabbitTemplate rabbitTemplate;
@RabbitListener(queues = DLQ)
public void rePublish(Message failedMessage) {
Map<String, Object> headers = failedMessage.getMessageProperties().getHeaders();
Integer retriesHeader = (Integer) headers.get(X_RETRIES_HEADER);
if (retriesHeader == null) {
retriesHeader = Integer.valueOf(0);
}
if (retriesHeader < 3) {
headers.put(X_RETRIES_HEADER, retriesHeader + 1);
String exchange = (String) headers.get(X_ORIGINAL_EXCHANGE_HEADER);
String originalRoutingKey = (String) headers.get(X_ORIGINAL_ROUTING_KEY_HEADER);
this.rabbitTemplate.send(exchange, originalRoutingKey, failedMessage);
}
else {
this.rabbitTemplate.send(PARKING_LOT, failedMessage);
}
}
@Bean
public Queue parkingLot() {
return new Queue(PARKING_LOT);
}
}
----

1361
docs/modules/ROOT/pages/rabbit/rabbit_overview.adoc Normal file
View File

File diff suppressed because it is too large Load Diff

128
docs/modules/ROOT/pages/rabbit/rabbit_partitions.adoc Normal file
View File

@@ -0,0 +1,128 @@
== Partitioning with the RabbitMQ Binder
RabbitMQ does not support partitioning natively.
Sometimes, it is advantageous to send data to specific partitions -- for example, when you want to strictly order message processing, all messages for a particular customer should go to the same partition.
The `RabbitMessageChannelBinder` provides partitioning by binding a queue for each partition to the destination exchange.
The following Java and YAML examples show how to configure the producer:
.Producer
[source, java]
----
@SpringBootApplication
public class RabbitPartitionProducerApplication {
private static final Random RANDOM = new Random(System.currentTimeMillis());
private static final String[] data = new String[] {
"abc1", "def1", "qux1",
"abc2", "def2", "qux2",
"abc3", "def3", "qux3",
"abc4", "def4", "qux4",
};
public static void main(String[] args) {
new SpringApplicationBuilder(RabbitPartitionProducerApplication.class)
.web(false)
.run(args);
}
@Bean
public Supplier<Message<?>> generate() {
return () -> {
String value = data[RANDOM.nextInt(data.length)];
System.out.println("Sending: " + value);
return MessageBuilder.withPayload(value)
.setHeader("partitionKey", value)
.build();
};
}
}
----
.application.yml
[source, yaml]
----
spring:
cloud:
stream:
bindings:
generate-out-0:
destination: partitioned.destination
producer:
partitioned: true
partition-key-expression: headers['partitionKey']
partition-count: 2
required-groups:
- myGroup
----
[NOTE]
====
The configuration in the prececing example uses the default partitioning (`key.hashCode() % partitionCount`).
This may or may not provide a suitably balanced algorithm, depending on the key values.
You can override this default by using the `partitionSelectorExpression` or `partitionSelectorClass` properties.
The `required-groups` property is required only if you need the consumer queues to be provisioned when the producer is deployed.
Otherwise, any messages sent to a partition are lost until the corresponding consumer is deployed.
====
The following configuration provisions a topic exchange:
image::part-exchange.png[scaledwidth="50%"]
The following queues are bound to that exchange:
image::part-queues.png[scaledwidth="50%"]
The following bindings associate the queues to the exchange:
image::part-bindings.png[scaledwidth="50%"]
The following Java and YAML examples continue the previous examples and show how to configure the consumer:
.Consumer
[source, java]
----
@SpringBootApplication
public class RabbitPartitionConsumerApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(RabbitPartitionConsumerApplication.class)
.web(false)
.run(args);
}
@Bean
public Consumer<Message<String>> listen() {
return message -> {
String queue =- message.getHeaders().get(AmqpHeaders.CONSUMER_QUEUE);
System.out.println(in + " received from queue " + queue);
};
}
}
----
.application.yml
[source, yaml]
----
spring:
cloud:
stream:
bindings:
listen-in-0:
destination: partitioned.destination
group: myGroup
consumer:
partitioned: true
instance-index: 0
----
IMPORTANT: The `RabbitMessageChannelBinder` does not support dynamic scaling.
There must be at least one consumer per partition.
The consumer's `instanceIndex` is used to indicate which partition is consumed.
Platforms such as Cloud Foundry can have only one instance with an `instanceIndex`.

44
docs/modules/ROOT/pages/rabbit/spring-cloud-stream-binder-rabbit.adoc Normal file
View File

@@ -0,0 +1,44 @@
:github-tag: master
:github-repo: spring-cloud/spring-cloud-stream-binder-rabbit
:github-raw: https://raw.githubusercontent.com/{github-repo}/{github-tag}
:github-code: https://github.com/{github-repo}/tree/{github-tag}
:toc: left
:toclevels: 8
:nofooter:
:sectlinks: true
[[spring-cloud-stream-binder-rabbit-reference]]
= Spring Cloud Stream RabbitMQ Binder Reference Guide
Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinathan, Gunnar Hillert, Mark Pollack, Patrick Peralta, Glenn Renfro, Thomas Risberg, Dave Syer, David Turanski, Janne Valkealahti, Benjamin Klein, Gary Russell, Jay Bryant
:doctype: book
:toc:
:toclevels: 4
:source-highlighter: prettify
:numbered:
:icons: font
:hide-uri-scheme:
:spring-cloud-stream-binder-rabbit-repo: snapshot
:github-tag: master
:spring-cloud-stream-binder-rabbit-docs-version: current
:spring-cloud-stream-binder-rabbit-docs: https://docs.spring.io/spring-cloud-stream-binder-rabbit/docs/{spring-cloud-stream-binder-rabbit-docs-version}/reference
:spring-cloud-stream-binder-rabbit-docs-current: https://docs.spring.io/spring-cloud-stream-binder-rabbit/docs/current-SNAPSHOT/reference/html/
:github-repo: spring-cloud/spring-cloud-stream-binder-rabbit
:github-raw: https://raw.github.com/{github-repo}/{github-tag}
:github-code: https://github.com/{github-repo}/tree/{github-tag}
:github-wiki: https://github.com/{github-repo}/wiki
:github-master-code: https://github.com/{github-repo}/tree/master
:sc-ext: java
// ======================================================================================
*{project-version}*
= Reference Guide
include::rabbit/rabbit_overview.adoc[]
include::rabbit/rabbit_dlq.adoc[]
include::rabbit/rabbit_partitions.adoc[]

26
docs/modules/ROOT/pages/sagan-index.adoc Normal file
View File

@@ -0,0 +1,26 @@
Spring Cloud Stream is a framework for building highly scalable event-driven microservices connected with shared messaging systems.
The framework provides a flexible programming model built on already established and familiar Spring idioms and best practices, including support
for persistent pub/sub semantics, consumer groups, and stateful partitions.
## Binder Implementations
Spring Cloud Stream supports a variety of binder implementations and the following table includes the link to the GitHub projects.
* https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit[RabbitMQ]
* https://github.com/spring-cloud/spring-cloud-stream-binder-kafka[Apache Kafka]
* https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/tree/master/spring-cloud-stream-binder-kafka-streams[Kafka Streams]
* https://github.com/spring-cloud/spring-cloud-stream-binder-aws-kinesis[Amazon Kinesis]
* https://github.com/GoogleCloudPlatform/spring-cloud-gcp/tree/main/spring-cloud-gcp-pubsub-stream-binder[Google PubSub _(partner maintained)_]
* https://github.com/SolaceProducts/solace-spring-cloud/tree/master/solace-spring-cloud-starters/solace-spring-cloud-stream-starter#spring-cloud-stream-binder-for-solace-pubsub[Solace PubSub+ _(partner maintained)_]
* https://aka.ms/spring/docs#spring-cloud-stream-binder-for-azure-event-hubs[Azure Event Hubs _(partner maintained)_]
* https://aka.ms/spring/docs#spring-cloud-stream-binder-for-azure-service-bus[Azure Service Bus _(partner maintained)_]
* https://github.com/alibaba/spring-cloud-alibaba/wiki/RocketMQ-en[Apache RocketMQ _(partner maintained)_]
The core building blocks of Spring Cloud Stream are:
* *Destination Binders*: Components responsible to provide integration with the external messaging systems.
* *Destination Bindings*: Bridge between the external messaging systems and application provided Producers and Consumers of messages (created by the Destination Binders).
* *Message*: The canonical data structure used by producers and consumers to communicate with Destination Binders (and thus other applications via external messaging systems).

372
docs/modules/ROOT/pages/schema-registry/spring-cloud-stream-schema-registry.adoc Normal file
View File

@@ -0,0 +1,372 @@
[[spring-cloud-stream-schema-registry-reference]]
= Spring Cloud Stream Schema Registry
== 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.
When such a schema needs to evolve to accommodate new business requirements, the existing components are still required to continue to work.
Spring Cloud Stream provides support for a standalone schema registry server using which aforementioned schema can be registered and used by the applications.
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 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.
Most serialization models, especially the ones that aim for portability across different platforms and languages, rely on a schema that describes how the data is serialized in the binary payload.
In order to serialize the data and then to interpret it, both the sending and receiving sides must have access to a schema that describes the binary format.
In certain cases, the schema can be inferred from the payload type on serialization or from the target type on deserialization.
However, many applications benefit from having access to an explicit schema that describes the binary data format.
A schema registry lets you store schema information in a textual format (typically JSON) and makes that information accessible to various applications that need it to receive and send data in binary format.
A schema is referenceable as a tuple consisting of:
* A subject that is the logical name of the schema
* The schema version
* The schema format, which describes the binary format of the data
Spring Cloud Stream Schema Registry provides the following components
* Standalone Schema Registry Server
By default, it is using an H2 database, but server can be used with PostgreSQL or MySQL by providing appropriate datasource configuration.
* Schema registry clients capable of message marshalling by communicating with a Schema Registry.
Currently, the client can communicate to the standalone schema registry or the Confluent Schema Registry.
=== Schema Registry Client
The client-side abstraction for interacting with schema registry servers is the `SchemaRegistryClient` interface, which has the following structure:
[source,java]
----
public interface SchemaRegistryClient {
SchemaRegistrationResponse register(String subject, String format, String schema);
String fetch(SchemaReference schemaReference);
String fetch(Integer id);
}
----
Spring Cloud Stream provides out-of-the-box implementations for interacting with its own schema server and for interacting with the Confluent Schema Registry.
A client for the Spring Cloud Stream schema registry can be configured by using the `@EnableSchemaRegistryClient`, as follows:
[source,java]
----
@SpringBootApplication
@EnableSchemaRegistryClient
public class ConsumerApplication {
}
----
NOTE: The default converter is optimized to cache not only the schemas from the remote server but also the `parse()` and `toString()` methods, which are quite expensive.
Because of this, it uses a `DefaultSchemaRegistryClient` that does not cache responses.
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
The Schema Registry Client supports the following properties:
`spring.cloud.stream.schemaRegistryClient.endpoint`:: The location of the schema-server.
When setting this, use a full URL, including protocol (`http` or `https`) , port, and context path.
+
Default:: `http://localhost:8990/`
`spring.cloud.stream.schemaRegistryClient.cached`:: Whether the client should cache schema server responses.
Normally set to `false`, as the caching happens in the message converter.
Clients using the schema registry client should set this to `true`.
+
Default:: `false`
=== 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.
This eases schema evolution, as applications that receive messages can get easy access to a writer schema that can be reconciled with their own reader schema.
For outbound messages, if the content type of the binding is set to `application/*+avro`, the `MessageConverter` is activated, as shown in the following example:
[source,properties]
----
spring.cloud.stream.stream.bindings.<output-binding-name>.contentType=application/*+avro
----
During the outbound conversion, the message converter tries to infer the schema of each outbound messages (based on its type) and register it to a subject (based on the payload type) by using the `SchemaRegistryClient`.
If an identical schema is already found, then a reference to it is retrieved.
If not, the schema is registered, and a new version number is provided.
The message is sent with a `contentType` header by using the following scheme: `application/[prefix].[subject].v[version]+avro`, where `prefix` is configurable and `subject` is deduced from the payload type.
For example, a message of the type `User` might be sent as a binary payload with a content type of `application/vnd.user.v2+avro`, where `user` is the subject and `2` is the version number.
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
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.
spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled:: Enable if you want the converter to use reflection to infer a Schema from a POJO.
+
Default: `false`
+
spring.cloud.stream.schema.avro.readerSchema:: Avro compares schema versions by looking at a writer schema (origin payload) and a reader schema (your application payload). See the https://avro.apache.org/docs/1.11.1/spec.html[Avro documentation] for more information.
If set, this overrides any lookups at the schema server and uses the local schema as the reader schema.
Default: `null`
+
spring.cloud.stream.schema.avro.schemaLocations:: Registers any `.avsc` files listed in this property with the Schema Server.
+
Default: `empty`
+
spring.cloud.stream.schema.avro.prefix:: The prefix to be used on the Content-Type header.
+
Default: `vnd`
spring.cloud.stream.schema.avro.subjectNamingStrategy:: Determines the subject name used to register the Avro schema in the schema registry. Two implementations are available, `org.springframework.cloud.stream.schema.avro.DefaultSubjectNamingStrategy`, where the subject is the schema name, and `org.springframework.cloud.stream.schema.avro.QualifiedSubjectNamingStrategy`, which returns a fully qualified subject using the Avro schema namespace and name. Custom strategies can be created by implementing `org.springframework.cloud.stream.schema.avro.SubjectNamingStrategy`.
+
Default: `org.springframework.cloud.stream.schema.avro.DefaultSubjectNamingStrategy`
+
spring.cloud.stream.schema.avro.ignoreSchemaRegistryServer:: Ignore any schema registry communication. Useful for testing purposes so that when running a unit test, it does not unnecessarily try to connect to a Schema Registry server.
+
Default: `false`
=== Apache Avro Message Converters
Spring Cloud Stream provides support for schema-based message converters through its `spring-cloud-stream-schema-registry-client` module.
Currently, the only serialization format supported out of the box for schema-based message converters is Apache Avro, with more formats to be added in future versions.
The `spring-cloud-stream-schema-registry-client` module contains two types of message converters that can be used for Apache Avro serialization:
* 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
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`).
If you provide a custom converter, then the default AvroSchemaMessageConverter bean is not created.
The following example shows a custom converter:
To use custom converters, you can simply add it to the application context, optionally specifying one or more `MimeTypes` with which to associate it.
The default `MimeType` is `application/avro`.
If the target type of the conversion is a `GenericRecord`, a schema must be set.
The following example shows how to configure a converter in a sink application by registering the Apache Avro `MessageConverter` without a predefined schema.
In this example, note that the mime type value is `avro/bytes`, not the default `application/avro`.
[source,java]
----
@SpringBootApplication
public static class SinkApplication {
//...
@Bean
public MessageConverter userMessageConverter() {
return new AvroSchemaMessageConverter(MimeType.valueOf("avro/bytes"));
}
}
----
Conversely, the following application registers a converter with a predefined schema (found on the classpath):
[source,java]
----
@SpringBootApplication
public static class SinkApplication {
//...
@Bean
public MessageConverter userMessageConverter() {
AvroSchemaMessageConverter converter = new AvroSchemaMessageConverter(MimeType.valueOf("avro/bytes"));
converter.setSchemaLocation(new ClassPathResource("schemas/User.avro"));
return converter;
}
}
----
=== Schema Registry Server
Spring Cloud Stream provides a schema registry server implementation.
To use it, you can download latest `spring-cloud-stream-schema-registry-server` release and run it as standalone application:
[source,bash]
----
wget https://repo1.maven.org/maven2/org/springframework/cloud/spring-cloud-stream-schema-registry-server/4.0.3/spring-cloud-stream-schema-registry-server-4.0.3.jar
java -jar ./spring-cloud-stream-schema-registry-server-4.0.3.jar
----
[TIP]
====
You can embed the schema registry in your existing Spring Boot web application.
To do this, add the `spring-cloud-stream-schema-registry-core` artifact to your project and use the `@EnableSchemaRegistryServer` annotation, which adds the schema registry server REST controller to your application.
The following example shows a Spring Boot application that enables the schema registry:
[source,java]
----
@SpringBootApplication
@EnableSchemaRegistryServer
public class SchemaRegistryServerApplication {
public static void main(String[] args) {
SpringApplication.run(SchemaRegistryServerApplication.class, args);
}
}
----
====
The `spring.cloud.stream.schema.server.path` property can be used to control the root path of the schema server (especially when it is embedded in other applications).
The `spring.cloud.stream.schema.server.allowSchemaDeletion` boolean property enables the deletion of a schema. By default, this is disabled.
The schema registry server uses a relational database to store the schemas.
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
The Schema Registry Server API consists of the following operations:
* `POST /` -- see `<<spring-cloud-stream-overview-registering-new-schema>>`
* `GET /{subject}/{format}/{version}` -- see `<<spring-cloud-stream-overview-retrieve-schema-subject-format-version>>`
* `GET /{subject}/{format}` -- see `<<spring-cloud-stream-overview-retrieve-schema-subject-format>>`
* `GET /schemas/{id}` -- see `<<spring-cloud-stream-overview-retrieve-schema-id>>`
* `DELETE /{subject}/{format}/{version}` -- see `<<spring-cloud-stream-overview-deleting-schema-subject-format-version>>`
* `DELETE /schemas/{id}` -- see `<<spring-cloud-stream-overview-deleting-schema-id>>`
* `DELETE /{subject}` -- see `<<spring-cloud-stream-overview-deleting-schema-subject>>`
[[spring-cloud-stream-overview-registering-new-schema]]
===== Registering a New Schema
To register a new schema, send a `POST` request to the `/` endpoint.
The `/` accepts a JSON payload with the following fields:
* `subject`: The schema subject
* `format`: The schema format
* `definition`: The schema definition
Its response is a schema object in JSON, with the following fields:
* `id`: The schema ID
* `subject`: The schema subject
* `format`: The schema format
* `version`: The schema version
* `definition`: The schema definition
[[spring-cloud-stream-overview-retrieve-schema-subject-format-version]]
===== Retrieving an Existing Schema by Subject, Format, and Version
To retrieve an existing schema by subject, format, and version, send `GET` request to the `/{subject}/{format}/{version}` endpoint.
Its response is a schema object in JSON, with the following fields:
* `id`: The schema ID
* `subject`: The schema subject
* `format`: The schema format
* `version`: The schema version
* `definition`: The schema definition
[[spring-cloud-stream-overview-retrieve-schema-subject-format]]
===== Retrieving an Existing Schema by Subject and Format
To retrieve an existing schema by subject and format, send a `GET` request to the `/subject/format` endpoint.
Its response is a list of schemas with each schema object in JSON, with the following fields:
* `id`: The schema ID
* `subject`: The schema subject
* `format`: The schema format
* `version`: The schema version
* `definition`: The schema definition
[[spring-cloud-stream-overview-retrieve-schema-id]]
===== Retrieving an Existing Schema by ID
To retrieve a schema by its ID, send a `GET` request to the `/schemas/{id}` endpoint.
Its response is a schema object in JSON, with the following fields:
* `id`: The schema ID
* `subject`: The schema subject
* `format`: The schema format
* `version`: The schema version
* `definition`: The schema definition
[[spring-cloud-stream-overview-deleting-schema-subject-format-version]]
===== Deleting a Schema by Subject, Format, and Version
To delete a schema identified by its subject, format, and version, send a `DELETE` request to the `/{subject}/{format}/{version}` endpoint.
[[spring-cloud-stream-overview-deleting-schema-id]]
===== Deleting a Schema by ID
To delete a schema by its ID, send a `DELETE` request to the `/schemas/{id}` endpoint.
[[spring-cloud-stream-overview-deleting-schema-subject]]
===== Deleting a Schema by Subject
`DELETE /{subject}`
Delete existing schemas by their subject.
NOTE: This note applies to users of Spring Cloud Stream 1.1.0.RELEASE only.
Spring Cloud Stream 1.1.0.RELEASE used the table name, `schema`, for storing `Schema` objects. `Schema` is a keyword in a number of database implementations.
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
The default configuration creates a `DefaultSchemaRegistryClient` bean.
If you want to use the Confluent schema registry, you need to create a bean of type `ConfluentSchemaRegistryClient`, which supersedes the one configured by default by the framework. The following example shows how to create such a bean:
[source,java]
----
@Bean
public SchemaRegistryClient schemaRegistryClient(@Value("${spring.cloud.stream.schemaRegistryClient.endpoint}") String endpoint){
ConfluentSchemaRegistryClient client = new ConfluentSchemaRegistryClient();
client.setEndpoint(endpoint);
return client;
}
----
NOTE: The ConfluentSchemaRegistryClient is tested against Confluent platform version 4.0.0.
=== 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:
* `<<spring-cloud-stream-overview-schema-registration-process>>`
* `<<spring-cloud-stream-overview-schema-resolution-process>>`
[[spring-cloud-stream-overview-schema-registration-process]]
==== Schema Registration Process (Serialization)
The first part of the registration process is extracting a schema from the payload that is being sent over a channel.
Avro types such as `SpecificRecord` or `GenericRecord` already contain a schema, which can be retrieved immediately from the instance.
In the case of POJOs, a schema is inferred if the `spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled` property is set to `true` (the default).
.Schema Writer Resolution Process
image::{github-raw}/docs/src/main/asciidoc/images/schema_resolution.png[width=800,scaledwidth="75%",align="center"]
Ones a schema is obtained, the converter loads its metadata (version) from the remote server.
First, it queries a local cache. If no result is found, it submits the data to the server, which replies with versioning information.
The converter always caches the results to avoid the overhead of querying the Schema Server for every new message that needs to be serialized.
.Schema Registration Process
image::{github-raw}/docs/src/main/asciidoc/images/registration.png[width=800,scaledwidth="75%",align="center"]
With the schema version information, the converter sets the `contentType` header of the message to carry the version information -- for example: `application/vnd.user.v1+avro`.
[[spring-cloud-stream-overview-schema-resolution-process]]
==== Schema Resolution Process (Deserialization)
When reading messages that contain version information (that is, a `contentType` header with a scheme like the one described under `<<spring-cloud-stream-overview-schema-registration-process>>`, the converter queries the Schema server to fetch the writer schema of the message.
Once it has found the correct schema of the incoming message, it retrieves the reader schema and, by using Avro's schema resolution support, reads it into the reader definition (setting defaults and any missing properties).
.Schema Reading Resolution Process
image::{github-raw}/docs/src/main/asciidoc/images/schema_reading.png[width=800,scaledwidth="75%",align="center"]
NOTE: You should understand the difference between a writer schema (the application that wrote the message) and a reader schema (the receiving application).
We suggest taking a moment to read https://avro.apache.org/docs/1.7.6/spec.html[the Avro terminology] and understand the process.
Spring Cloud Stream always fetches the writer schema to determine how to read a message.
If you want to get Avro's schema evolution support working, you need to make sure that a `readerSchema` was properly set for your application.

3258
docs/modules/ROOT/pages/spring-cloud-stream.adoc Normal file
View File

File diff suppressed because it is too large Load Diff