224 lines
12 KiB
Plaintext
224 lines
12 KiB
Plaintext
[[reactive-kafka-binder]]
|
|
= Reactive Kafka Binder
|
|
|
|
Kafka binder ecosystem in Spring Cloud Stream provides a dedicated reactive binder based on the https://projectreactor.io/docs/kafka/release/reference/[Reactor Kafka] project.
|
|
This reactive Kafka binder enables full end-to-end reactive capabilities such as backpressure, reactive streams etc. in applications based on Apache Kafka.
|
|
When your Spring Cloud Stream Kafka application is written using reactive types (`Flux`, `Mono` etc.), it is recommended to use this reactive Kafka binder instead of the regular message channel based Kafka binder.
|
|
|
|
[[maven-coordinates]]
|
|
== Maven Coordinates
|
|
|
|
Following are the maven coordinates for the reactive Kafka binder.
|
|
|
|
```
|
|
<dependency>
|
|
<groupId>org.springframework.cloud</groupId>
|
|
<artifactId>spring-cloud-stream-binder-kafka-reactive</artifactId>
|
|
</dependency>
|
|
```
|
|
|
|
[[basic-example-using-the-reactive-kafka-binder]]
|
|
== Basic Example using the Reactive Kafka Binder
|
|
|
|
In this section, we show some basic code snippets for writing a reactive Kafka application using the reactive binder and details around them.
|
|
|
|
[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]]
|
|
== 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]]
|
|
== 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]]
|
|
== Multiplex
|
|
|
|
Starting with version 4.0.3, the common consumer property `multiplex` is now supported by the reactive binder, where a single binding can consume from multiple topics.
|
|
When `false` (default), a separate binding is created for each topic specified in a comma-delimited list in the common `destination` property.
|
|
|
|
[[destination-is-pattern]]
|
|
== Destination is Pattern
|
|
|
|
Starting with version 4.0.3, the `destination-is-pattern` Kafka binding consumer property is now supported.
|
|
The receiver options are conigured with a regex `Pattern`, allowing the binding to consume from any topic that matches the pattern.
|
|
|
|
[[sender-result-channel]]
|
|
== Sender Result Channel
|
|
|
|
Starting with version 4.0.3, you can configure the `resultMetadataChannel` to receive `SenderResult<?>` s to determine success/failure of sends.
|
|
|
|
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.
|