Splits Reactive Support doc into multiple documents (#631)
Resolves #463
This commit is contained in:
ka_sh
@@ -14,6 +14,12 @@
|
||||
*** xref:reference/pulsar/publishing-consuming-partitioned-topics.adoc[]
|
||||
*** xref:reference/tombstones.adoc[]
|
||||
** xref:reference/reactive-pulsar.adoc[]
|
||||
*** xref:reference/reactive-pulsar/reactive-quick-tour.adoc[]
|
||||
*** xref:reference/reactive-pulsar/reactive-design.adoc[]
|
||||
*** xref:reference/reactive-pulsar/reactive-pulsar-client.adoc[]
|
||||
*** xref:reference/reactive-pulsar/reactive-message-production.adoc[]
|
||||
*** xref:reference/reactive-pulsar/reactive-message-consumption.adoc[]
|
||||
*** xref:reference/reactive-pulsar/reactive-topic-resolution.adoc[]
|
||||
*** xref:reference/tombstones-reactive.adoc[]
|
||||
** xref:reference/pulsar-admin.adoc[]
|
||||
** xref:reference/pulsar-function.adoc[]
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[[reactive-pulsar]]
|
||||
= Reactive Support
|
||||
include::../attributes/attributes.adoc[]
|
||||
:page-section-summary-toc: 1
|
||||
|
||||
The framework provides a Reactive counterpart for almost all supported features.
|
||||
|
||||
@@ -27,475 +27,3 @@ To do so, you can add the `spring-pulsar-reactive-spring-boot-starter` module as
|
||||
|
||||
NOTE: The majority of this reference expects the reader to be using the starter and gives most directions for configuration with that in mind.
|
||||
However, an effort is made to call out when instructions are specific to the Spring Boot starter usage.
|
||||
|
||||
include::reactive-quick-tour.adoc[leveloffset=+1]
|
||||
|
||||
== Design
|
||||
|
||||
Here are a few key design points to keep in mind.
|
||||
|
||||
=== Apache Pulsar Reactive
|
||||
The reactive support is ultimately provided by the https://github.com/apache/pulsar-client-reactive[Apache Pulsar Reactive client] whose current implementation is a fully non-blocking adapter around the regular Pulsar client's asynchronous API.
|
||||
This implies that the Reactive client requires the regular client.
|
||||
|
||||
=== Additive Auto-Configuration
|
||||
Due to the dependence on the regular (imperative) client, the Reactive auto-configuration provided by the framework is additive to the imperative auto-configuration.
|
||||
In other words, The imperative starter only includes the imperative components but the reactive starter includes both imperative and reactive components.
|
||||
|
||||
[[reactive-pulsar-client]]
|
||||
== Reactive Pulsar Client
|
||||
When you use the Reactive Pulsar Spring Boot Starter, you get the `ReactivePulsarClient` auto-configured.
|
||||
|
||||
By default, the application tries to connect to a local Pulsar instance at `pulsar://localhost:6650`.
|
||||
This can be adjusted by setting the `spring.pulsar.client.service-url` property to a different value.
|
||||
|
||||
TIP: The value must be a valid {apache-pulsar-docs}/client-libraries-java/#connection-urls[Pulsar Protocol] URL
|
||||
|
||||
There are many other application properties (inherited from the adapted imperative client) available to configure.
|
||||
See the {spring-boot-pulsar-config-props}[`spring.pulsar.client.*`] application properties.
|
||||
|
||||
[[reactive-client-authentication]]
|
||||
=== Authentication
|
||||
To connect to a Pulsar cluster that requires authentication, follow xref:reference/pulsar/pulsar-client.adoc#client-authentication[the same steps] as the imperative client.
|
||||
Again, this is because the reactive client adapts the imperative client which handles all security configuration.
|
||||
|
||||
[[reactive-message-production]]
|
||||
== Message Production
|
||||
|
||||
[[reactive-pulsar-template]]
|
||||
=== ReactivePulsarTemplate
|
||||
On the Pulsar producer side, Spring Boot auto-configuration provides a `ReactivePulsarTemplate` for publishing records. The template implements an interface called `ReactivePulsarOperations` and provides methods to publish records through its contract.
|
||||
|
||||
The template provides send methods that accept a single message and return a `Mono<MessageId>`.
|
||||
It also provides send methods that accept multiple messages (in the form of the ReactiveStreams `Publisher` type) and return a `Flux<MessageId>`.
|
||||
|
||||
NOTE: For the API variants that do not include a topic parameter, a <<topic-resolution-process-reactive,topic resolution process>> is used to determine the destination topic.
|
||||
|
||||
==== Fluent API
|
||||
The template provides a {javadocs}/org/springframework/pulsar/reactive/core/ReactivePulsarOperations.html#newMessage(T)[fluent builder] to handle more complicated send requests.
|
||||
|
||||
==== Message customization
|
||||
You can specify a `MessageSpecBuilderCustomizer` to configure the outgoing message. For example, the following code shows how to send a keyed message:
|
||||
[source, java]
|
||||
----
|
||||
template.newMessage(msg)
|
||||
.withMessageCustomizer((mc) -> mc.key("foo-msg-key"))
|
||||
.send();
|
||||
----
|
||||
|
||||
==== Sender customization
|
||||
You can specify a `ReactiveMessageSenderBuilderCustomizer` to configure the underlying Pulsar sender builder that ultimately constructs the sender used to send the outgoing message.
|
||||
|
||||
WARNING: Use with caution as this gives full access to the sender builder and invoking some of its methods (such as `create`) may have unintended side effects.
|
||||
|
||||
For example, the following code shows how to disable batching and enable chunking:
|
||||
[source, java]
|
||||
----
|
||||
template.newMessage(msg)
|
||||
.withSenderCustomizer((sc) -> sc.enableChunking(true).enableBatching(false))
|
||||
.send();
|
||||
----
|
||||
|
||||
This other example shows how to use custom routing when publishing records to partitioned topics.
|
||||
Specify your custom `MessageRouter` implementation on the sender builder such as:
|
||||
[source, java]
|
||||
----
|
||||
template.newMessage(msg)
|
||||
.withSenderCustomizer((sc) -> sc.messageRouter(messageRouter))
|
||||
.send();
|
||||
----
|
||||
|
||||
TIP: Note that, when using a `MessageRouter`, the only valid setting for `spring.pulsar.producer.message-routing-mode` is `custom`.
|
||||
|
||||
[[schema-info-template-reactive]]
|
||||
:template-class: ReactivePulsarTemplate
|
||||
include::schema-info/schema-info-template.adoc[leveloffset=+1]
|
||||
|
||||
[[reactive-sender-factory]]
|
||||
=== ReactivePulsarSenderFactory
|
||||
The `ReactivePulsarTemplate` relies on a `ReactivePulsarSenderFactory` to actually create the underlying sender.
|
||||
|
||||
Spring Boot provides this sender factory which can be configured with any of the {spring-boot-pulsar-config-props}[`spring.pulsar.producer.*`] application properties.
|
||||
|
||||
NOTE: If topic information is not specified when using the sender factory APIs directly, the same <<topic-resolution-process-reactive,topic resolution process>> used by the `ReactivePulsarTemplate` is used with the one exception that the "Message type default" step is **omitted**.
|
||||
|
||||
==== Producer Caching
|
||||
Each underlying Pulsar producer consumes resources.
|
||||
To improve performance and avoid continual creation of producers, the `ReactiveMessageSenderCache` in the underlying Apache Pulsar Reactive client caches the producers that it creates.
|
||||
They are cached in an LRU fashion and evicted when they have not been used within a configured time period.
|
||||
|
||||
You can configure the cache settings by specifying any of the {spring-boot-pulsar-config-props}[`spring.pulsar.producer.cache.*`] application properties.
|
||||
|
||||
[[reactive-message-consumption]]
|
||||
== Message Consumption
|
||||
|
||||
[[reactive-pulsar-listener]]
|
||||
=== @ReactivePulsarListener
|
||||
|
||||
When it comes to Pulsar consumers, we recommend that end-user applications use the `ReactivePulsarListener` annotation.
|
||||
To use `ReactivePulsarListener`, you need to use the `@EnableReactivePulsar` annotation.
|
||||
When you use Spring Boot support, it automatically enables this annotation and configures all necessary components, such as the message listener infrastructure (which is responsible for creating the underlying Pulsar consumer).
|
||||
|
||||
Let us revisit the `ReactivePulsarListener` code snippet we saw in the quick-tour section:
|
||||
|
||||
[source, java]
|
||||
----
|
||||
@ReactivePulsarListener(subscriptionName = "hello-pulsar-sub", topics = "hello-pulsar-topic")
|
||||
Mono<Void> listen(String message) {
|
||||
System.out.println(message);
|
||||
return Mono.empty();
|
||||
}
|
||||
----
|
||||
|
||||
NOTE: The listener method returns a `Mono<Void>` to signal whether the message was successfully processed. `Mono.empty()` indicates success (acknowledgment) and `Mono.error()` indicates failure (negative acknowledgment).
|
||||
|
||||
You can also further simplify this method:
|
||||
[source, java]
|
||||
----
|
||||
@ReactivePulsarListener
|
||||
Mono<Void> listen(String message) {
|
||||
System.out.println(message);
|
||||
return Mono.empty();
|
||||
}
|
||||
----
|
||||
|
||||
In this most basic form, when the `topics` are not directly provided, a <<topic-resolution-process-reactive,topic resolution process>> is used to determine the destination topic.
|
||||
Likewise, when the `subscriptionName` is not provided on the `@ReactivePulsarListener` annotation an auto-generated subscription name will be used.
|
||||
|
||||
In the `ReactivePulsarListener` method shown earlier, we receive the data as `String`, but we do not specify any schema types.
|
||||
Internally, the framework relies on Pulsar's schema mechanism to convert the data to the required type.
|
||||
|
||||
The framework detects that you expect the `String` type and then infers the schema type based on that information and provides that schema to the consumer.
|
||||
The framework does this inference for all primitive types.
|
||||
For all non-primitive types the default schema is assumed to be JSON.
|
||||
If a complex type is using anything besides JSON (such as AVRO or KEY_VALUE) you must provide the schema type on the annotation using the `schemaType` property.
|
||||
|
||||
This example shows how we can consume complex types from a topic:
|
||||
[source, java]
|
||||
----
|
||||
@ReactivePulsarListener(topics = "my-topic-2", schemaType = SchemaType.JSON)
|
||||
Mono<Void> listen(Foo message) {
|
||||
System.out.println(message);
|
||||
return Mono.empty();
|
||||
}
|
||||
----
|
||||
|
||||
Let us look at a few more ways we can consume.
|
||||
|
||||
This example consumes the Pulsar message directly:
|
||||
[source, java]
|
||||
----
|
||||
@ReactivePulsarListener(topics = "my-topic")
|
||||
Mono<Void> listen(org.apache.pulsar.client.api.Message<String> message) {
|
||||
System.out.println(message.getValue());
|
||||
return Mono.empty();
|
||||
}
|
||||
----
|
||||
|
||||
This example consumes the record wrapped in a Spring messaging envelope:
|
||||
[source, java]
|
||||
----
|
||||
@ReactivePulsarListener(topics = "my-topic")
|
||||
Mono<Void> listen(org.springframework.messaging.Message<String> message) {
|
||||
System.out.println(message.getPayload());
|
||||
return Mono.empty();
|
||||
}
|
||||
----
|
||||
|
||||
==== Streaming
|
||||
All of the above are examples of consuming a single record one-by-one.
|
||||
However, one of the compelling reasons to use Reactive is for the streaming capability with backpressure support.
|
||||
|
||||
The following example uses `ReactivePulsarListener` to consume a stream of POJOs:
|
||||
|
||||
[source, java]
|
||||
----
|
||||
@ReactivePulsarListener(topics = "streaming-1", stream = true)
|
||||
Flux<MessageResult<Void>> listen(Flux<org.apache.pulsar.client.api.Message<String>> messages) {
|
||||
return messages
|
||||
.doOnNext((msg) -> System.out.println("Received: " + msg.getValue()))
|
||||
.map(MessageResult::acknowledge);
|
||||
----
|
||||
Here we receive the records as a `Flux` of Pulsar messages.
|
||||
In addition, to enable stream consumption at the `ReactivePulsarListener` level, you need to set the `stream` property on the annotation to `true`.
|
||||
|
||||
NOTE: The listener method returns a `Flux<MessageResult<Void>>` where each element represents a processed message and holds the message id, value and whether it was acknowledged.
|
||||
The `MessageResult` has a set of static factory methods that can be used to create the appropriate `MessageResult` instance.
|
||||
|
||||
Based on the actual type of the messages in the `Flux`, the framework tries to infer the schema to use.
|
||||
If it contains a complex type, you still need to provide the `schemaType` on `ReactivePulsarListener`.
|
||||
|
||||
The following listener uses the Spring messaging `Message` envelope with a complex type :
|
||||
[source, java]
|
||||
----
|
||||
@ReactivePulsarListener(topics = "streaming-2", stream = true, schemaType = SchemaType.JSON)
|
||||
Flux<MessageResult<Void>> listen2(Flux<org.springframework.messaging.Message<Foo>> messages) {
|
||||
return messages
|
||||
.doOnNext((msg) -> System.out.println("Received: " + msg.getPayload()))
|
||||
.map(MessageUtils::acknowledge);
|
||||
}
|
||||
----
|
||||
|
||||
NOTE: The listener method returns a `Flux<MessageResult<Void>>` where each element represents a processed message and holds the message id, value and whether it was acknowledged.
|
||||
The Spring `MessageUtils` has a set of static factory methods that can be used to create the appropriate `MessageResult` instance from a Spring message.
|
||||
|
||||
==== Configuration - Application Properties
|
||||
The listener relies on the `ReactivePulsarConsumerFactory` to create and manage the underlying Pulsar consumer that it uses to consume messages.
|
||||
Spring Boot provides this consumer factory which you can further configure by specifying the {spring-boot-pulsar-config-props}[`spring.pulsar.consumer.*`] application properties.
|
||||
**Most** of the configured properties on the factory will be respected in the listener with the following **exceptions**:
|
||||
|
||||
TIP: The `spring.pulsar.consumer.subscription.name` property is ignored and is instead generated when not specified on the annotation.
|
||||
|
||||
TIP: The `spring.pulsar.consumer.subscription-type` property is ignored and is instead taken from the value on the annotation. However, you can set the `subscriptionType = {}` on the annotation to instead use the property value as the default.
|
||||
|
||||
==== Generic records with AUTO_CONSUME
|
||||
If there is no chance to know the type of schema of a Pulsar topic in advance, you can use the `AUTO_CONSUME` schema type to consume generic records.
|
||||
In this case, the topic deserializes messages into `GenericRecord` objects using the schema info associated with the topic.
|
||||
|
||||
To consume generic records set the `schemaType = SchemaType.AUTO_CONSUME` on your `@ReactivePulsarListener` and use a Pulsar message of type `GenericRecord` as the message parameter as shown below.
|
||||
|
||||
[source, java]
|
||||
----
|
||||
@ReactivePulsarListener(topics = "my-generic-topic", schemaType = SchemaType.AUTO_CONSUME)
|
||||
Mono<Void> listen(org.apache.pulsar.client.api.Message<GenericRecord> message) {
|
||||
GenericRecord record = message.getValue();
|
||||
record.getFields().forEach((f) ->
|
||||
System.out.printf("%s = %s%n", f.getName(), record.getField(f)));
|
||||
return Mono.empty();
|
||||
}
|
||||
----
|
||||
|
||||
TIP: The `GenericRecord` API allows access to the fields and their associated values
|
||||
|
||||
|
||||
[[reactive-consumer-customizer]]
|
||||
==== Consumer Customization
|
||||
|
||||
You can specify a `ReactivePulsarListenerMessageConsumerBuilderCustomizer` to configure the underlying Pulsar consumer builder that ultimately constructs the consumer used by the listener to receive the messages.
|
||||
|
||||
WARNING: Use with caution as this gives full access to the consumer builder and invoking some of its methods (such as `create`) may have unintended side effects.
|
||||
|
||||
For example, the following code shows how to set the initial position of the subscription to the earliest messaage on the topic.
|
||||
|
||||
[source, java]
|
||||
----
|
||||
@ReactivePulsarListener(topics = "hello-pulsar-topic", consumerCustomizer = "myConsumerCustomizer")
|
||||
Mono<Void> listen(String message) {
|
||||
System.out.println(message);
|
||||
return Mono.empty();
|
||||
}
|
||||
|
||||
@Bean
|
||||
ReactivePulsarListenerMessageConsumerBuilderCustomizer<String> myConsumerCustomizer() {
|
||||
return b -> b.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
|
||||
}
|
||||
----
|
||||
|
||||
TIP: If your application only has a single `@ReactivePulsarListener` and a single `ReactivePulsarListenerMessageConsumerBuilderCustomizer` bean registered then the customizer will be automatically applied.
|
||||
|
||||
You can also use the customizer to provide direct Pulsar consumer properties to the consumer builder.
|
||||
This is convenient if you do not want to use the Boot configuration properties mentioned earlier or have multiple `ReactivePulsarListener` methods whose configuration varies.
|
||||
|
||||
The following customizer example uses direct Pulsar consumer properties:
|
||||
|
||||
[source, java]
|
||||
----
|
||||
@Bean
|
||||
ReactivePulsarListenerMessageConsumerBuilderCustomizer<String> directConsumerPropsCustomizer() {
|
||||
return b -> b.property("subscriptionName", "subscription-1").property("topicNames", "foo-1");
|
||||
}
|
||||
----
|
||||
|
||||
CAUTION: The properties used are direct Pulsar consumer properties, not the `spring.pulsar.consumer` Spring Boot configuration properties
|
||||
|
||||
[[schema-info-listener-reactive]]
|
||||
:listener-class: ReactivePulsarListener
|
||||
include::schema-info/schema-info-listener.adoc[leveloffset=+1]
|
||||
|
||||
[[reactive-message-listener-container]]
|
||||
=== Message Listener Container Infrastructure
|
||||
|
||||
In most scenarios, we recommend using the `ReactivePulsarListener` annotation directly for consuming from a Pulsar topic as that model covers a broad set of application use cases.
|
||||
However, it is important to understand how `ReactivePulsarListener` works internally.
|
||||
|
||||
The message listener container is at the heart of message consumption when you use Spring for Apache Pulsar.
|
||||
The `ReactivePulsarListener` uses the message listener container infrastructure behind the scenes to create and manage the underlying Pulsar consumer.
|
||||
|
||||
==== ReactivePulsarMessageListenerContainer
|
||||
The contract for this message listener container is provided through `ReactivePulsarMessageListenerContainer` whose default implementation creates a reactive Pulsar consumer and wires up a reactive message pipeline that uses the created consumer.
|
||||
|
||||
==== ReactiveMessagePipeline
|
||||
The pipeline is a feature of the underlying Apache Pulsar Reactive client which does the heavy lifting of receiving the data in a reactive manner and then handing it over to the provided message handler. The reactive message listener container implementation is much simpler because the pipeline handles the majority of the work.
|
||||
|
||||
==== ReactivePulsarMessageHandler
|
||||
The "listener" aspect is provided by the `ReactivePulsarMessageHandler` of which there are two provided implementations:
|
||||
|
||||
* `ReactivePulsarOneByOneMessageHandler` - handles a single message one-by-one
|
||||
* `ReactivePulsarStreamingHandler` - handles multiple messages via a `Flux`
|
||||
|
||||
NOTE: If topic information is not specified when using the listener containers directly, the same <<topic-resolution-process-reactive,topic resolution process>> used by the `ReactivePulsarListener` is used with the one exception that the "Message type default" step is **omitted**.
|
||||
|
||||
[[reactive-concurrency]]
|
||||
=== Concurrency
|
||||
When consuming records in streaming mode (`stream = true`) concurrency comes naturally via the underlying Reactive support in the client implementation.
|
||||
|
||||
However, when handling messages one-by-one, concurrency can be specified to increase processing throughput.
|
||||
Simply set the `concurrency` property on `@ReactivePulsarListener`.
|
||||
Additionally, when `concurrency > 1` you can ensure messages are ordered by key and therefore sent to the same handler by setting `useKeyOrderedProcessing = "true"` on the annotation.
|
||||
|
||||
Again, the `ReactiveMessagePipeline` does the heavy lifting, we simply set the properties on it.
|
||||
|
||||
.[small]#Reactive vs Imperative#
|
||||
****
|
||||
Concurrency in the reactive container is different from its imperative counterpart.
|
||||
The latter creates multiple threads (each with a Pulsar consumer) whereas the former dispatches the messages to multiple handler instances concurrently on the Reactive parallel scheduler.
|
||||
|
||||
One advantage of the reactive concurrency model is that it can be used with `Exclusive` subscriptions whereas the imperative concurrency model can not.
|
||||
****
|
||||
|
||||
[[reactive-pulsar-headers]]
|
||||
=== Pulsar Headers
|
||||
The Pulsar message metadata can be consumed as Spring message headers.
|
||||
The list of available headers can be found in {github}/blob/main/spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarHeaders.java[PulsarHeaders.java].
|
||||
|
||||
[[reactive-pulsar-headers.single]]
|
||||
==== Accessing In OneByOne Listener
|
||||
The following example shows how you can access Pulsar Headers when using a one-by-one message listener:
|
||||
|
||||
[source,java]
|
||||
----
|
||||
@ReactivePulsarListener(topics = "some-topic")
|
||||
Mono<Void> listen(String data,
|
||||
@Header(PulsarHeaders.MESSAGE_ID) MessageId messageId,
|
||||
@Header("foo") String foo) {
|
||||
System.out.println("Received " + data + " w/ id=" + messageId + " w/ foo=" + foo);
|
||||
return Mono.empty();
|
||||
}
|
||||
----
|
||||
|
||||
In the preceding example, we access the values for the `messageId` message metadata as well as a custom message property named `foo`.
|
||||
The Spring `@Header` annotation is used for each header field.
|
||||
|
||||
You can also use Pulsar's `Message` as the envelope to carry the payload.
|
||||
When doing so, the user can directly call the corresponding methods on the Pulsar message for retrieving the metadata.
|
||||
However, as a convenience, you can also retrieve it by using the `Header` annotation.
|
||||
Note that you can also use the Spring messaging `Message` envelope to carry the payload and then retrieve the Pulsar headers by using `@Header`.
|
||||
|
||||
[[reactive-pulsar-headers.streaming]]
|
||||
==== Accessing In Streaming Listener
|
||||
When using a streaming message listener the header support is limited.
|
||||
Only when the `Flux` contains Spring `org.springframework.messaging.Message` elements will the headers be populated.
|
||||
Additionally, the Spring `@Header` annotation can not be used to retrieve the data.
|
||||
You must directly call the corresponding methods on the Spring message to retrieve the data.
|
||||
|
||||
|
||||
[[reactive-message-ack]]
|
||||
=== Message Acknowledgment
|
||||
The framework automatically handles message acknowledgement.
|
||||
However, the listener method must send a signal indicating whether the message was successfully processed.
|
||||
The container implementation then uses that signal to perform the ack or nack operation.
|
||||
This is a slightly different from its imperative counterpart where the signal is implied as positive unless the method throws an exception.
|
||||
|
||||
==== OneByOne Listener
|
||||
The single message (aka OneByOne) message listener method returns a `Mono<Void>` to signal whether the message was successfully processed. `Mono.empty()` indicates success (acknowledgment) and `Mono.error()` indicates failure (negative acknowledgment).
|
||||
|
||||
==== Streaming Listener
|
||||
The streaming listener method returns a `Flux<MessageResult<Void>>` where each `MessageResult` element represents a processed message and holds the message id, value and whether it was acknowledged. The `MessageResult` has a set of `acknowledge` and `negativeAcknowledge` static factory methods that can be used to create the appropriate `MessageResult` instance.
|
||||
|
||||
[[reactive-redelivery]]
|
||||
=== Message Redelivery and Error Handling
|
||||
Apache Pulsar provides various native strategies for message redelivery and error handling.
|
||||
We will take a look at them and see how to use them through Spring for Apache Pulsar.
|
||||
|
||||
==== Acknowledgment Timeout
|
||||
|
||||
By default, Pulsar consumers do not redeliver messages unless the consumer crashes, but you can change this behavior by setting an ack timeout on the Pulsar consumer.
|
||||
If the ack timeout property has a value above zero and if the Pulsar consumer does not acknowledge a message within that timeout period, the message is redelivered.
|
||||
|
||||
You can specify this property directly as a Pulsar consumer property via a <<reactive-consumer-customizer,consumer customizer>> such as:
|
||||
|
||||
[source, java]
|
||||
----
|
||||
@Bean
|
||||
ReactiveMessageConsumerBuilderCustomizer<String> consumerCustomizer() {
|
||||
return b -> b.property("ackTimeout", "60s");
|
||||
}
|
||||
----
|
||||
|
||||
==== Negative Acknowledgment Redelivery Delay
|
||||
|
||||
When acknowledging negatively, Pulsar consumer lets you specify how the application wants the message to be re-delivered.
|
||||
The default is to redeliver the message in one minute, but you can change it via a <<reactive-consumer-customizer,consumer customizer>> such as:
|
||||
|
||||
[source, java]
|
||||
----
|
||||
@Bean
|
||||
ReactiveMessageConsumerBuilderCustomizer<String> consumerCustomizer() {
|
||||
return b -> b.property("negativeAckRedeliveryDelay", "10ms");
|
||||
}
|
||||
----
|
||||
|
||||
==== Dead Letter Topic
|
||||
Apache Pulsar lets applications use a dead letter topic on consumers with a `Shared` subscription type.
|
||||
For the `Exclusive` and `Failover` subscription types, this feature is not available.
|
||||
The basic idea is that, if a message is retried a certain number of times (maybe due to an ack timeout or nack redelivery), once the number of retries are exhausted, the message can be sent to a special topic called the dead letter queue (DLQ).
|
||||
Let us see some details around this feature in action by inspecting some code snippets:
|
||||
|
||||
[source, java]
|
||||
----
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
class DeadLetterPolicyConfig {
|
||||
|
||||
@ReactivePulsarListener(
|
||||
topics = "topic-with-dlp",
|
||||
subscriptionType = SubscriptionType.Shared,
|
||||
deadLetterPolicy = "myDeadLetterPolicy",
|
||||
consumerCustomizer = "ackTimeoutCustomizer" )
|
||||
void listen(String msg) {
|
||||
throw new RuntimeException("fail " + msg);
|
||||
}
|
||||
|
||||
@ReactivePulsarListener(topics = "my-dlq-topic")
|
||||
void listenDlq(String msg) {
|
||||
System.out.println("From DLQ: " + msg);
|
||||
}
|
||||
|
||||
@Bean
|
||||
DeadLetterPolicy myDeadLetterPolicy() {
|
||||
return DeadLetterPolicy.builder().maxRedeliverCount(10).deadLetterTopic("my-dlq-topic").build();
|
||||
}
|
||||
|
||||
@Bean
|
||||
ReactiveMessageConsumerBuilderCustomizer<String> ackTimeoutCustomizer() {
|
||||
return b -> b.property("ackTimeout", "1s");
|
||||
}
|
||||
}
|
||||
----
|
||||
|
||||
First, we have a special bean for `DeadLetterPolicy`, and it is named as `deadLetterPolicy` (it can be any name as you wish).
|
||||
This bean specifies a number of things, such as the max delivery (10, in this case) and the name of the dead letter topic -- `my-dlq-topic`, in this case.
|
||||
If you do not specify a DLQ topic name, it defaults to `<topicname>-<subscriptionname>-DLQ` in Pulsar.
|
||||
Next, we provide this bean name to `ReactivePulsarListener` by setting the `deadLetterPolicy` property.
|
||||
Note that the `ReactivePulsarListener` has a subscription type of `Shared`, as the DLQ feature only works with shared subscriptions.
|
||||
This code is primarily for demonstration purposes, so we provide an `ackTimeout` value of 1 second.
|
||||
The idea is that the code throws the exception and, if Pulsar does not receive an ack within 1 second, it does a retry.
|
||||
If that cycle continues ten times (as that is our max redelivery count in the `DeadLetterPolicy`), the Pulsar consumer publishes the messages to the DLQ topic.
|
||||
We have another `ReactivePulsarListener` that listens on the DLQ topic to receive data as it is published to the DLQ topic.
|
||||
|
||||
.Special note on DLQ topics when using partitioned topics
|
||||
****
|
||||
If the main topic is partitioned, behind the scenes, each partition is treated as a separate topic by Pulsar.
|
||||
Pulsar appends `partition-<n>`, where `n` stands for the partition number to the main topic name.
|
||||
The problem is that, if you do not specify a DLQ topic (as opposed to what we did above), Pulsar publishes to a default topic name that has this ``partition-<n>` info in it -- for example: `topic-with-dlp-partition-0-deadLetterPolicySubscription-DLQ`.
|
||||
The easy way to solve this is to provide a DLQ topic name always.
|
||||
****
|
||||
|
||||
[[reactive-pulsar-reader]]
|
||||
=== Pulsar Reader Support
|
||||
The framework provides support for using {apache-pulsar-docs}/concepts-clients/#reader-interface[Pulsar Reader] in a Reactive fashion via the `ReactivePulsarReaderFactory`.
|
||||
|
||||
Spring Boot provides this reader factory which can be configured with any of the {spring-boot-pulsar-config-props}[`spring.pulsar.reader.*`] application properties.
|
||||
|
||||
[[topic-resolution-process-reactive]]
|
||||
== Topic Resolution
|
||||
include::pulsar/topic-resolution.adoc[leveloffset=+1]
|
||||
|
||||
@@ -0,0 +1,13 @@
|
||||
[[reactive-design]]
|
||||
= Design
|
||||
include::../../attributes/attributes.adoc[]
|
||||
|
||||
Here are a few key design points to keep in mind.
|
||||
|
||||
== Apache Pulsar Reactive
|
||||
The reactive support is ultimately provided by the https://github.com/apache/pulsar-client-reactive[Apache Pulsar Reactive client] whose current implementation is a fully non-blocking adapter around the regular Pulsar client's asynchronous API.
|
||||
This implies that the Reactive client requires the regular client.
|
||||
|
||||
== Additive Auto-Configuration
|
||||
Due to the dependence on the regular (imperative) client, the Reactive auto-configuration provided by the framework is additive to the imperative auto-configuration.
|
||||
In other words, The imperative starter only includes the imperative components but the reactive starter includes both imperative and reactive components.
|
||||
@@ -0,0 +1,370 @@
|
||||
[[reactive-message-consumption]]
|
||||
= Message Consumption
|
||||
include::../../attributes/attributes.adoc[]
|
||||
|
||||
[[reactive-pulsar-listener]]
|
||||
== @ReactivePulsarListener
|
||||
|
||||
When it comes to Pulsar consumers, we recommend that end-user applications use the `ReactivePulsarListener` annotation.
|
||||
To use `ReactivePulsarListener`, you need to use the `@EnableReactivePulsar` annotation.
|
||||
When you use Spring Boot support, it automatically enables this annotation and configures all necessary components, such as the message listener infrastructure (which is responsible for creating the underlying Pulsar consumer).
|
||||
|
||||
Let us revisit the `ReactivePulsarListener` code snippet we saw in the quick-tour section:
|
||||
|
||||
[source, java]
|
||||
----
|
||||
@ReactivePulsarListener(subscriptionName = "hello-pulsar-sub", topics = "hello-pulsar-topic")
|
||||
Mono<Void> listen(String message) {
|
||||
System.out.println(message);
|
||||
return Mono.empty();
|
||||
}
|
||||
----
|
||||
|
||||
NOTE: The listener method returns a `Mono<Void>` to signal whether the message was successfully processed. `Mono.empty()` indicates success (acknowledgment) and `Mono.error()` indicates failure (negative acknowledgment).
|
||||
|
||||
You can also further simplify this method:
|
||||
[source, java]
|
||||
----
|
||||
@ReactivePulsarListener
|
||||
Mono<Void> listen(String message) {
|
||||
System.out.println(message);
|
||||
return Mono.empty();
|
||||
}
|
||||
----
|
||||
|
||||
In this most basic form, when the `topics` are not directly provided, a xref:reference/reactive-pulsar/reactive-topic-resolution.adoc#topic-resolution-process-reactive[topic resolution process] is used to determine the destination topic.
|
||||
Likewise, when the `subscriptionName` is not provided on the `@ReactivePulsarListener` annotation an auto-generated subscription name will be used.
|
||||
|
||||
In the `ReactivePulsarListener` method shown earlier, we receive the data as `String`, but we do not specify any schema types.
|
||||
Internally, the framework relies on Pulsar's schema mechanism to convert the data to the required type.
|
||||
|
||||
The framework detects that you expect the `String` type and then infers the schema type based on that information and provides that schema to the consumer.
|
||||
The framework does this inference for all primitive types.
|
||||
For all non-primitive types the default schema is assumed to be JSON.
|
||||
If a complex type is using anything besides JSON (such as AVRO or KEY_VALUE) you must provide the schema type on the annotation using the `schemaType` property.
|
||||
|
||||
This example shows how we can consume complex types from a topic:
|
||||
[source, java]
|
||||
----
|
||||
@ReactivePulsarListener(topics = "my-topic-2", schemaType = SchemaType.JSON)
|
||||
Mono<Void> listen(Foo message) {
|
||||
System.out.println(message);
|
||||
return Mono.empty();
|
||||
}
|
||||
----
|
||||
|
||||
Let us look at a few more ways we can consume.
|
||||
|
||||
This example consumes the Pulsar message directly:
|
||||
[source, java]
|
||||
----
|
||||
@ReactivePulsarListener(topics = "my-topic")
|
||||
Mono<Void> listen(org.apache.pulsar.client.api.Message<String> message) {
|
||||
System.out.println(message.getValue());
|
||||
return Mono.empty();
|
||||
}
|
||||
----
|
||||
|
||||
This example consumes the record wrapped in a Spring messaging envelope:
|
||||
[source, java]
|
||||
----
|
||||
@ReactivePulsarListener(topics = "my-topic")
|
||||
Mono<Void> listen(org.springframework.messaging.Message<String> message) {
|
||||
System.out.println(message.getPayload());
|
||||
return Mono.empty();
|
||||
}
|
||||
----
|
||||
|
||||
=== Streaming
|
||||
All of the above are examples of consuming a single record one-by-one.
|
||||
However, one of the compelling reasons to use Reactive is for the streaming capability with backpressure support.
|
||||
|
||||
The following example uses `ReactivePulsarListener` to consume a stream of POJOs:
|
||||
|
||||
[source, java]
|
||||
----
|
||||
@ReactivePulsarListener(topics = "streaming-1", stream = true)
|
||||
Flux<MessageResult<Void>> listen(Flux<org.apache.pulsar.client.api.Message<String>> messages) {
|
||||
return messages
|
||||
.doOnNext((msg) -> System.out.println("Received: " + msg.getValue()))
|
||||
.map(MessageResult::acknowledge);
|
||||
----
|
||||
Here we receive the records as a `Flux` of Pulsar messages.
|
||||
In addition, to enable stream consumption at the `ReactivePulsarListener` level, you need to set the `stream` property on the annotation to `true`.
|
||||
|
||||
NOTE: The listener method returns a `Flux<MessageResult<Void>>` where each element represents a processed message and holds the message id, value and whether it was acknowledged.
|
||||
The `MessageResult` has a set of static factory methods that can be used to create the appropriate `MessageResult` instance.
|
||||
|
||||
Based on the actual type of the messages in the `Flux`, the framework tries to infer the schema to use.
|
||||
If it contains a complex type, you still need to provide the `schemaType` on `ReactivePulsarListener`.
|
||||
|
||||
The following listener uses the Spring messaging `Message` envelope with a complex type :
|
||||
[source, java]
|
||||
----
|
||||
@ReactivePulsarListener(topics = "streaming-2", stream = true, schemaType = SchemaType.JSON)
|
||||
Flux<MessageResult<Void>> listen2(Flux<org.springframework.messaging.Message<Foo>> messages) {
|
||||
return messages
|
||||
.doOnNext((msg) -> System.out.println("Received: " + msg.getPayload()))
|
||||
.map(MessageUtils::acknowledge);
|
||||
}
|
||||
----
|
||||
|
||||
NOTE: The listener method returns a `Flux<MessageResult<Void>>` where each element represents a processed message and holds the message id, value and whether it was acknowledged.
|
||||
The Spring `MessageUtils` has a set of static factory methods that can be used to create the appropriate `MessageResult` instance from a Spring message.
|
||||
|
||||
=== Configuration - Application Properties
|
||||
The listener relies on the `ReactivePulsarConsumerFactory` to create and manage the underlying Pulsar consumer that it uses to consume messages.
|
||||
Spring Boot provides this consumer factory which you can further configure by specifying the {spring-boot-pulsar-config-props}[`spring.pulsar.consumer.*`] application properties.
|
||||
**Most** of the configured properties on the factory will be respected in the listener with the following **exceptions**:
|
||||
|
||||
TIP: The `spring.pulsar.consumer.subscription.name` property is ignored and is instead generated when not specified on the annotation.
|
||||
|
||||
TIP: The `spring.pulsar.consumer.subscription-type` property is ignored and is instead taken from the value on the annotation. However, you can set the `subscriptionType = {}` on the annotation to instead use the property value as the default.
|
||||
|
||||
=== Generic records with AUTO_CONSUME
|
||||
If there is no chance to know the type of schema of a Pulsar topic in advance, you can use the `AUTO_CONSUME` schema type to consume generic records.
|
||||
In this case, the topic deserializes messages into `GenericRecord` objects using the schema info associated with the topic.
|
||||
|
||||
To consume generic records set the `schemaType = SchemaType.AUTO_CONSUME` on your `@ReactivePulsarListener` and use a Pulsar message of type `GenericRecord` as the message parameter as shown below.
|
||||
|
||||
[source, java]
|
||||
----
|
||||
@ReactivePulsarListener(topics = "my-generic-topic", schemaType = SchemaType.AUTO_CONSUME)
|
||||
Mono<Void> listen(org.apache.pulsar.client.api.Message<GenericRecord> message) {
|
||||
GenericRecord record = message.getValue();
|
||||
record.getFields().forEach((f) ->
|
||||
System.out.printf("%s = %s%n", f.getName(), record.getField(f)));
|
||||
return Mono.empty();
|
||||
}
|
||||
----
|
||||
|
||||
TIP: The `GenericRecord` API allows access to the fields and their associated values
|
||||
|
||||
|
||||
[[reactive-consumer-customizer]]
|
||||
=== Consumer Customization
|
||||
|
||||
You can specify a `ReactivePulsarListenerMessageConsumerBuilderCustomizer` to configure the underlying Pulsar consumer builder that ultimately constructs the consumer used by the listener to receive the messages.
|
||||
|
||||
WARNING: Use with caution as this gives full access to the consumer builder and invoking some of its methods (such as `create`) may have unintended side effects.
|
||||
|
||||
For example, the following code shows how to set the initial position of the subscription to the earliest messaage on the topic.
|
||||
|
||||
[source, java]
|
||||
----
|
||||
@ReactivePulsarListener(topics = "hello-pulsar-topic", consumerCustomizer = "myConsumerCustomizer")
|
||||
Mono<Void> listen(String message) {
|
||||
System.out.println(message);
|
||||
return Mono.empty();
|
||||
}
|
||||
|
||||
@Bean
|
||||
ReactivePulsarListenerMessageConsumerBuilderCustomizer<String> myConsumerCustomizer() {
|
||||
return b -> b.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
|
||||
}
|
||||
----
|
||||
|
||||
TIP: If your application only has a single `@ReactivePulsarListener` and a single `ReactivePulsarListenerMessageConsumerBuilderCustomizer` bean registered then the customizer will be automatically applied.
|
||||
|
||||
You can also use the customizer to provide direct Pulsar consumer properties to the consumer builder.
|
||||
This is convenient if you do not want to use the Boot configuration properties mentioned earlier or have multiple `ReactivePulsarListener` methods whose configuration varies.
|
||||
|
||||
The following customizer example uses direct Pulsar consumer properties:
|
||||
|
||||
[source, java]
|
||||
----
|
||||
@Bean
|
||||
ReactivePulsarListenerMessageConsumerBuilderCustomizer<String> directConsumerPropsCustomizer() {
|
||||
return b -> b.property("subscriptionName", "subscription-1").property("topicNames", "foo-1");
|
||||
}
|
||||
----
|
||||
|
||||
CAUTION: The properties used are direct Pulsar consumer properties, not the `spring.pulsar.consumer` Spring Boot configuration properties
|
||||
|
||||
[[schema-info-listener-reactive]]
|
||||
:listener-class: ReactivePulsarListener
|
||||
include::../schema-info/schema-info-listener.adoc[]
|
||||
|
||||
[[reactive-message-listener-container]]
|
||||
== Message Listener Container Infrastructure
|
||||
|
||||
In most scenarios, we recommend using the `ReactivePulsarListener` annotation directly for consuming from a Pulsar topic as that model covers a broad set of application use cases.
|
||||
However, it is important to understand how `ReactivePulsarListener` works internally.
|
||||
|
||||
The message listener container is at the heart of message consumption when you use Spring for Apache Pulsar.
|
||||
The `ReactivePulsarListener` uses the message listener container infrastructure behind the scenes to create and manage the underlying Pulsar consumer.
|
||||
|
||||
=== ReactivePulsarMessageListenerContainer
|
||||
The contract for this message listener container is provided through `ReactivePulsarMessageListenerContainer` whose default implementation creates a reactive Pulsar consumer and wires up a reactive message pipeline that uses the created consumer.
|
||||
|
||||
=== ReactiveMessagePipeline
|
||||
The pipeline is a feature of the underlying Apache Pulsar Reactive client which does the heavy lifting of receiving the data in a reactive manner and then handing it over to the provided message handler. The reactive message listener container implementation is much simpler because the pipeline handles the majority of the work.
|
||||
|
||||
=== ReactivePulsarMessageHandler
|
||||
The "listener" aspect is provided by the `ReactivePulsarMessageHandler` of which there are two provided implementations:
|
||||
|
||||
* `ReactivePulsarOneByOneMessageHandler` - handles a single message one-by-one
|
||||
* `ReactivePulsarStreamingHandler` - handles multiple messages via a `Flux`
|
||||
|
||||
NOTE: If topic information is not specified when using the listener containers directly, the same xref:reference/reactive-pulsar/reactive-topic-resolution.adoc#topic-resolution-process-reactive[topic resolution process] used by the `ReactivePulsarListener` is used with the one exception that the "Message type default" step is **omitted**.
|
||||
|
||||
[[reactive-concurrency]]
|
||||
== Concurrency
|
||||
When consuming records in streaming mode (`stream = true`) concurrency comes naturally via the underlying Reactive support in the client implementation.
|
||||
|
||||
However, when handling messages one-by-one, concurrency can be specified to increase processing throughput.
|
||||
Simply set the `concurrency` property on `@ReactivePulsarListener`.
|
||||
Additionally, when `concurrency > 1` you can ensure messages are ordered by key and therefore sent to the same handler by setting `useKeyOrderedProcessing = "true"` on the annotation.
|
||||
|
||||
Again, the `ReactiveMessagePipeline` does the heavy lifting, we simply set the properties on it.
|
||||
|
||||
.[small]#Reactive vs Imperative#
|
||||
****
|
||||
Concurrency in the reactive container is different from its imperative counterpart.
|
||||
The latter creates multiple threads (each with a Pulsar consumer) whereas the former dispatches the messages to multiple handler instances concurrently on the Reactive parallel scheduler.
|
||||
|
||||
One advantage of the reactive concurrency model is that it can be used with `Exclusive` subscriptions whereas the imperative concurrency model can not.
|
||||
****
|
||||
|
||||
[[reactive-pulsar-headers]]
|
||||
== Pulsar Headers
|
||||
The Pulsar message metadata can be consumed as Spring message headers.
|
||||
The list of available headers can be found in {github}/blob/main/spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarHeaders.java[PulsarHeaders.java].
|
||||
|
||||
[[reactive-pulsar-headers.single]]
|
||||
=== Accessing In OneByOne Listener
|
||||
The following example shows how you can access Pulsar Headers when using a one-by-one message listener:
|
||||
|
||||
[source,java]
|
||||
----
|
||||
@ReactivePulsarListener(topics = "some-topic")
|
||||
Mono<Void> listen(String data,
|
||||
@Header(PulsarHeaders.MESSAGE_ID) MessageId messageId,
|
||||
@Header("foo") String foo) {
|
||||
System.out.println("Received " + data + " w/ id=" + messageId + " w/ foo=" + foo);
|
||||
return Mono.empty();
|
||||
}
|
||||
----
|
||||
|
||||
In the preceding example, we access the values for the `messageId` message metadata as well as a custom message property named `foo`.
|
||||
The Spring `@Header` annotation is used for each header field.
|
||||
|
||||
You can also use Pulsar's `Message` as the envelope to carry the payload.
|
||||
When doing so, the user can directly call the corresponding methods on the Pulsar message for retrieving the metadata.
|
||||
However, as a convenience, you can also retrieve it by using the `Header` annotation.
|
||||
Note that you can also use the Spring messaging `Message` envelope to carry the payload and then retrieve the Pulsar headers by using `@Header`.
|
||||
|
||||
[[reactive-pulsar-headers.streaming]]
|
||||
=== Accessing In Streaming Listener
|
||||
When using a streaming message listener the header support is limited.
|
||||
Only when the `Flux` contains Spring `org.springframework.messaging.Message` elements will the headers be populated.
|
||||
Additionally, the Spring `@Header` annotation can not be used to retrieve the data.
|
||||
You must directly call the corresponding methods on the Spring message to retrieve the data.
|
||||
|
||||
|
||||
[[reactive-message-ack]]
|
||||
== Message Acknowledgment
|
||||
The framework automatically handles message acknowledgement.
|
||||
However, the listener method must send a signal indicating whether the message was successfully processed.
|
||||
The container implementation then uses that signal to perform the ack or nack operation.
|
||||
This is a slightly different from its imperative counterpart where the signal is implied as positive unless the method throws an exception.
|
||||
|
||||
=== OneByOne Listener
|
||||
The single message (aka OneByOne) message listener method returns a `Mono<Void>` to signal whether the message was successfully processed. `Mono.empty()` indicates success (acknowledgment) and `Mono.error()` indicates failure (negative acknowledgment).
|
||||
|
||||
=== Streaming Listener
|
||||
The streaming listener method returns a `Flux<MessageResult<Void>>` where each `MessageResult` element represents a processed message and holds the message id, value and whether it was acknowledged. The `MessageResult` has a set of `acknowledge` and `negativeAcknowledge` static factory methods that can be used to create the appropriate `MessageResult` instance.
|
||||
|
||||
[[reactive-redelivery]]
|
||||
== Message Redelivery and Error Handling
|
||||
Apache Pulsar provides various native strategies for message redelivery and error handling.
|
||||
We will take a look at them and see how to use them through Spring for Apache Pulsar.
|
||||
|
||||
=== Acknowledgment Timeout
|
||||
|
||||
By default, Pulsar consumers do not redeliver messages unless the consumer crashes, but you can change this behavior by setting an ack timeout on the Pulsar consumer.
|
||||
If the ack timeout property has a value above zero and if the Pulsar consumer does not acknowledge a message within that timeout period, the message is redelivered.
|
||||
|
||||
You can specify this property directly as a Pulsar consumer property via a <<reactive-consumer-customizer,consumer customizer>> such as:
|
||||
|
||||
[source, java]
|
||||
----
|
||||
@Bean
|
||||
ReactiveMessageConsumerBuilderCustomizer<String> consumerCustomizer() {
|
||||
return b -> b.property("ackTimeout", "60s");
|
||||
}
|
||||
----
|
||||
|
||||
=== Negative Acknowledgment Redelivery Delay
|
||||
|
||||
When acknowledging negatively, Pulsar consumer lets you specify how the application wants the message to be re-delivered.
|
||||
The default is to redeliver the message in one minute, but you can change it via a <<reactive-consumer-customizer,consumer customizer>> such as:
|
||||
|
||||
[source, java]
|
||||
----
|
||||
@Bean
|
||||
ReactiveMessageConsumerBuilderCustomizer<String> consumerCustomizer() {
|
||||
return b -> b.property("negativeAckRedeliveryDelay", "10ms");
|
||||
}
|
||||
----
|
||||
|
||||
=== Dead Letter Topic
|
||||
Apache Pulsar lets applications use a dead letter topic on consumers with a `Shared` subscription type.
|
||||
For the `Exclusive` and `Failover` subscription types, this feature is not available.
|
||||
The basic idea is that, if a message is retried a certain number of times (maybe due to an ack timeout or nack redelivery), once the number of retries are exhausted, the message can be sent to a special topic called the dead letter queue (DLQ).
|
||||
Let us see some details around this feature in action by inspecting some code snippets:
|
||||
|
||||
[source, java]
|
||||
----
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
class DeadLetterPolicyConfig {
|
||||
|
||||
@ReactivePulsarListener(
|
||||
topics = "topic-with-dlp",
|
||||
subscriptionType = SubscriptionType.Shared,
|
||||
deadLetterPolicy = "myDeadLetterPolicy",
|
||||
consumerCustomizer = "ackTimeoutCustomizer" )
|
||||
void listen(String msg) {
|
||||
throw new RuntimeException("fail " + msg);
|
||||
}
|
||||
|
||||
@ReactivePulsarListener(topics = "my-dlq-topic")
|
||||
void listenDlq(String msg) {
|
||||
System.out.println("From DLQ: " + msg);
|
||||
}
|
||||
|
||||
@Bean
|
||||
DeadLetterPolicy myDeadLetterPolicy() {
|
||||
return DeadLetterPolicy.builder().maxRedeliverCount(10).deadLetterTopic("my-dlq-topic").build();
|
||||
}
|
||||
|
||||
@Bean
|
||||
ReactiveMessageConsumerBuilderCustomizer<String> ackTimeoutCustomizer() {
|
||||
return b -> b.property("ackTimeout", "1s");
|
||||
}
|
||||
}
|
||||
----
|
||||
|
||||
First, we have a special bean for `DeadLetterPolicy`, and it is named as `deadLetterPolicy` (it can be any name as you wish).
|
||||
This bean specifies a number of things, such as the max delivery (10, in this case) and the name of the dead letter topic -- `my-dlq-topic`, in this case.
|
||||
If you do not specify a DLQ topic name, it defaults to `<topicname>-<subscriptionname>-DLQ` in Pulsar.
|
||||
Next, we provide this bean name to `ReactivePulsarListener` by setting the `deadLetterPolicy` property.
|
||||
Note that the `ReactivePulsarListener` has a subscription type of `Shared`, as the DLQ feature only works with shared subscriptions.
|
||||
This code is primarily for demonstration purposes, so we provide an `ackTimeout` value of 1 second.
|
||||
The idea is that the code throws the exception and, if Pulsar does not receive an ack within 1 second, it does a retry.
|
||||
If that cycle continues ten times (as that is our max redelivery count in the `DeadLetterPolicy`), the Pulsar consumer publishes the messages to the DLQ topic.
|
||||
We have another `ReactivePulsarListener` that listens on the DLQ topic to receive data as it is published to the DLQ topic.
|
||||
|
||||
.Special note on DLQ topics when using partitioned topics
|
||||
****
|
||||
If the main topic is partitioned, behind the scenes, each partition is treated as a separate topic by Pulsar.
|
||||
Pulsar appends `partition-<n>`, where `n` stands for the partition number to the main topic name.
|
||||
The problem is that, if you do not specify a DLQ topic (as opposed to what we did above), Pulsar publishes to a default topic name that has this ``partition-<n>` info in it -- for example: `topic-with-dlp-partition-0-deadLetterPolicySubscription-DLQ`.
|
||||
The easy way to solve this is to provide a DLQ topic name always.
|
||||
****
|
||||
|
||||
[[reactive-pulsar-reader]]
|
||||
== Pulsar Reader Support
|
||||
The framework provides support for using {apache-pulsar-docs}/concepts-clients/#reader-interface[Pulsar Reader] in a Reactive fashion via the `ReactivePulsarReaderFactory`.
|
||||
|
||||
Spring Boot provides this reader factory which can be configured with any of the {spring-boot-pulsar-config-props}[`spring.pulsar.reader.*`] application properties.
|
||||
@@ -0,0 +1,67 @@
|
||||
[[reactive-message-production]]
|
||||
= Message Production
|
||||
include::../../attributes/attributes.adoc[]
|
||||
|
||||
[[reactive-pulsar-template]]
|
||||
== ReactivePulsarTemplate
|
||||
On the Pulsar producer side, Spring Boot auto-configuration provides a `ReactivePulsarTemplate` for publishing records. The template implements an interface called `ReactivePulsarOperations` and provides methods to publish records through its contract.
|
||||
|
||||
The template provides send methods that accept a single message and return a `Mono<MessageId>`.
|
||||
It also provides send methods that accept multiple messages (in the form of the ReactiveStreams `Publisher` type) and return a `Flux<MessageId>`.
|
||||
|
||||
NOTE: For the API variants that do not include a topic parameter, a xref:reference/reactive-pulsar/reactive-topic-resolution.adoc#topic-resolution-process-reactive[topic resolution process] is used to determine the destination topic.
|
||||
|
||||
=== Fluent API
|
||||
The template provides a {javadocs}/org/springframework/pulsar/reactive/core/ReactivePulsarOperations.html#newMessage(T)[fluent builder] to handle more complicated send requests.
|
||||
|
||||
=== Message customization
|
||||
You can specify a `MessageSpecBuilderCustomizer` to configure the outgoing message. For example, the following code shows how to send a keyed message:
|
||||
[source, java]
|
||||
----
|
||||
template.newMessage(msg)
|
||||
.withMessageCustomizer((mc) -> mc.key("foo-msg-key"))
|
||||
.send();
|
||||
----
|
||||
|
||||
=== Sender customization
|
||||
You can specify a `ReactiveMessageSenderBuilderCustomizer` to configure the underlying Pulsar sender builder that ultimately constructs the sender used to send the outgoing message.
|
||||
|
||||
WARNING: Use with caution as this gives full access to the sender builder and invoking some of its methods (such as `create`) may have unintended side effects.
|
||||
|
||||
For example, the following code shows how to disable batching and enable chunking:
|
||||
[source, java]
|
||||
----
|
||||
template.newMessage(msg)
|
||||
.withSenderCustomizer((sc) -> sc.enableChunking(true).enableBatching(false))
|
||||
.send();
|
||||
----
|
||||
|
||||
This other example shows how to use custom routing when publishing records to partitioned topics.
|
||||
Specify your custom `MessageRouter` implementation on the sender builder such as:
|
||||
[source, java]
|
||||
----
|
||||
template.newMessage(msg)
|
||||
.withSenderCustomizer((sc) -> sc.messageRouter(messageRouter))
|
||||
.send();
|
||||
----
|
||||
|
||||
TIP: Note that, when using a `MessageRouter`, the only valid setting for `spring.pulsar.producer.message-routing-mode` is `custom`.
|
||||
|
||||
[[schema-info-template-reactive]]
|
||||
:template-class: ReactivePulsarTemplate
|
||||
include::../schema-info/schema-info-template.adoc[]
|
||||
|
||||
[[reactive-sender-factory]]
|
||||
== ReactivePulsarSenderFactory
|
||||
The `ReactivePulsarTemplate` relies on a `ReactivePulsarSenderFactory` to actually create the underlying sender.
|
||||
|
||||
Spring Boot provides this sender factory which can be configured with any of the {spring-boot-pulsar-config-props}[`spring.pulsar.producer.*`] application properties.
|
||||
|
||||
NOTE: If topic information is not specified when using the sender factory APIs directly, the same xref:reference/reactive-pulsar/reactive-topic-resolution.adoc#topic-resolution-process-reactive[topic resolution process] used by the `ReactivePulsarTemplate` is used with the one exception that the "Message type default" step is **omitted**.
|
||||
|
||||
=== Producer Caching
|
||||
Each underlying Pulsar producer consumes resources.
|
||||
To improve performance and avoid continual creation of producers, the `ReactiveMessageSenderCache` in the underlying Apache Pulsar Reactive client caches the producers that it creates.
|
||||
They are cached in an LRU fashion and evicted when they have not been used within a configured time period.
|
||||
|
||||
You can configure the cache settings by specifying any of the {spring-boot-pulsar-config-props}[`spring.pulsar.producer.cache.*`] application properties.
|
||||
@@ -0,0 +1,18 @@
|
||||
[[reactive-pulsar-client]]
|
||||
= Reactive Pulsar Client
|
||||
include::../../attributes/attributes.adoc[]
|
||||
|
||||
When you use the Reactive Pulsar Spring Boot Starter, you get the `ReactivePulsarClient` auto-configured.
|
||||
|
||||
By default, the application tries to connect to a local Pulsar instance at `pulsar://localhost:6650`.
|
||||
This can be adjusted by setting the `spring.pulsar.client.service-url` property to a different value.
|
||||
|
||||
TIP: The value must be a valid {apache-pulsar-docs}/client-libraries-java/#connection-urls[Pulsar Protocol] URL
|
||||
|
||||
There are many other application properties (inherited from the adapted imperative client) available to configure.
|
||||
See the {spring-boot-pulsar-config-props}[`spring.pulsar.client.*`] application properties.
|
||||
|
||||
[[reactive-client-authentication]]
|
||||
== Authentication
|
||||
To connect to a Pulsar cluster that requires authentication, follow xref:reference/pulsar/pulsar-client.adoc#client-authentication[the same steps] as the imperative client.
|
||||
Again, this is because the reactive client adapts the imperative client which handles all security configuration.
|
||||
@@ -1,6 +1,6 @@
|
||||
[[quick-tour-reactive]]
|
||||
= Quick Tour
|
||||
include::../attributes/attributes.adoc[]
|
||||
include::../../attributes/attributes.adoc[]
|
||||
|
||||
We will take a quick tour of the Reactive support in Spring for Apache Pulsar by showing a sample Spring Boot application that produces and consumes in a Reactive fashion.
|
||||
This is a complete application and does not require any additional configuration, as long as you have a Pulsar cluster running on the default location - `localhost:6650`.
|
||||
@@ -0,0 +1,3 @@
|
||||
[[topic-resolution-process-reactive]]
|
||||
= Topic Resolution
|
||||
include::../pulsar/topic-resolution.adoc[]
|
||||
Reference in New Issue
Block a user