Spring/spring-pulsar
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Clarify use of default subscription name in @PulsarListener (#482)

Browse Source 
* With the move to Spring Boot the configuration properties were
synthesized (imperative + reactive) into single properties. This results in 
property path changes and this commit adjusts for that.

* Update docs to clarify Reactive vs. Imperative concurrency models

See #480
This commit is contained in:
Chris Bono
2023-11-11 00:02:13 -06:00
committed by GitHub
parent 728a74bcb0
commit f5582df022
2 changed files with 25 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar.adoc
View File

@@ -151,7 +151,9 @@ To use `PulsarListener`, you need to use the `@EnablePulsar` annotation.
When you use Spring Boot support, it automatically enables this annotation and configures all the components necessary for `PulsarListener`, such as the message listener infrastructure (which is responsible for creating the Pulsar consumer).
`PulsarMessageListenerContainer` uses a `PulsarConsumerFactory` to create and manage the Pulsar consumer.
Spring Boot auto-configuration also provides this consumer factory which you can further configure by specifying any of the {spring-boot-pulsar-config-props}[`spring.pulsar.consumer.*`] application properties.
Spring Boot auto-configuration also provides this consumer factory which you can further configure by specifying **most** of the {spring-boot-pulsar-config-props}[`spring.pulsar.consumer.*`] application properties.
NOTE: `spring.pulsar.consumer.subscription.name` is ignored and is instead generated when not specified on the annotation.
Let us revisit the `PulsarListener` code snippet we saw in the quick-tour section:
@@ -173,17 +175,8 @@ public void listen(String message) {
}
----
In this most basic form, you must provide the following two properties with their corresponding values:
[source,yaml,indent=0,subs="verbatim"]
----
spring.pulsar.consumer:
topics: hello-pulsar
subscription-name: hello-pulsar-subscription
----
NOTE: If the topic information is not directly provided, a <<topic-resolution-process-imperative,topic resolution process>> is used to determine the destination topic.
In this most basic form, when the `subscriptionName` is not provided on the `@PulsarListener` annotation an auto-generated subscription name will be used.
Likewise, when the `topics` are not directly provided, a <<topic-resolution-process-imperative,topic resolution process>> is used to determine the destination topic.
In the `PulsarListener` 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.
@@ -593,10 +586,6 @@ public void listen(String message) {
}
----
You can also set the listener property, `spring.pulsar.listner.ack-mode`, to set the ack mode application-wide.
When doing this, you need not set this on the `PulsarListener` annotation.
In that case, all the `PulsarListener` methods in the application acquire that property.
==== Manual Message Ack in Single Record Mode
You might not always want the framework to send acknowledgments but, rather, do that directly from the application itself.
@@ -738,11 +727,10 @@ Apache Pulsar provides various native strategies for message redelivery and erro
==== Specifying Acknowledgment Timeout for Message Redelivery
By default, Pulsar consumers does not redeliver messages unless the consumer crashes, but you can change this behavior by setting an ack timeout on the Pulsar consumer.
When you use Spring for Apache Pulsar, you can enable this property by setting the `spring.pulsar.consumer.ack-timeout` Boot property.
If this property has a value above zero and if the Pulsar consumer does not acknowledge a message within that timeout period, the message is redelivered.
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 also specify this property directly as a Pulsar consumer property on the `PulsarListener` itself:
When you use Spring for Apache Pulsar, you can set this property via a <<_consumer_customization_on_pulsarlistener,consumer customizer>> or with the native Pulsar `ackTimeout` property in the `properties` attribute of `@PulsarListener`:
[source, java]
----
@@ -753,7 +741,7 @@ public void listen(String s) {
}
----
When you specify `ackTimeout` (as seen in the preceding `PulsarListener` method), if the consumer does not send an acknowledgement within 60 seconds, the message is redelivered by Pulsar to the consumer.
When you specify the ack timeout, if the consumer does not send an acknowledgement within 60 seconds, the message is redelivered by Pulsar to the consumer.
If you want to specify some advanced backoff options for ack timeout with different delays, you can do the following:
@@ -787,8 +775,7 @@ We provide the backoff bean to the `PulsarListener` annotation by setting the `a
==== Specifying Negative Acknowledgment Redelivery
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 by setting `spring.pulsar.consumer.negative-ack-redelivery-delay`.
You can also set it as a consumer property directly on `PulsarListener`, as follows:
The default is to redeliver the message in one minute, but you can change it via a <<_consumer_customization_on_pulsarlistener,consumer customizer>> or with the native Pulsar `negativeAckRedeliveryDelay` property in the `properties` attribute of `@PulsarListener`:
[source, java]
----

42
spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar.adoc
View File

@@ -105,7 +105,7 @@ template.newMessage(msg)
.send();
----
TIP: Note that, when using a `MessageRouter`, the only valid setting for `spring.pulsar.reactive.sender.message-routing-mode` is `custom`.
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
@@ -115,7 +115,7 @@ include::schema-info/schema-info-template.adoc[leveloffset=+1]
=== 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.reactive.sender.*`] application properties.
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**.
@@ -124,7 +124,7 @@ 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.reactive.sender.cache.*`] application properties.
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
@@ -159,18 +159,8 @@ Mono<Void> listen(String message) {
}
----
In this most basic form, you must still provide the topic name by setting the following property:
[source,yaml,indent=0,subs="verbatim"]
----
spring.pulsar.reactive.consumer:
topic-names: hello-pulsar-topic
----
When `subscription-name` is not provided an auto-generated subscription name will be used.
NOTE: If the topic information is not directly provided, a <<topic-resolution-process-reactive,topic resolution process>> is used to determine the destination topic.
In this most basic form, when the `subscriptionName` is not provided on the `@ReactivePulsarListener` annotation an auto-generated subscription name will be used.
Likewise, when the `topics` are not directly provided, a <<topic-resolution-process-reactive,topic resolution process>> is used to determine the destination topic.
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.
@@ -250,7 +240,7 @@ Flux<MessageResult<Void>> listen2(Flux<org.springframework.messaging.Message<Foo
==== Configuration - Application Properties
The listener ultimately relies on `ReactivePulsarConsumerFactory` to create and manage the underlying Pulsar consumer.
Spring Boot provides this consumer factory which can be configured with any of the {spring-boot-pulsar-config-props}[`spring.pulsar.reactive.consumer.*`] application-properties.
Spring Boot provides this consumer factory which can be configured with any of the {spring-boot-pulsar-config-props}[`spring.pulsar.consumer.*`] application-properties.
[[reactive-consumer-customizer]]
==== Consumer Customization
@@ -288,7 +278,7 @@ ReactiveMessageConsumerBuilderCustomizer<String> directConsumerPropsCustomizer()
}
----
CAUTION: The properties used are direct Pulsar consumer properties, not the `spring.pulsar.reactive.consumer` Spring Boot configuration properties
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
@@ -332,8 +322,7 @@ Again, the `ReactiveMessagePipeline` does the heavy lifting, we simply set the p
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 reactive concurrency is that it can be used with `Exclusive` and `Failover` subscriptions to increase processing throughput if strict ordering is not required.
In contrast to imperative concurrency that can not currently be used with `Exclusive` and does not provide more processing power with `Failover`.
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]]
@@ -389,11 +378,11 @@ Apache Pulsar provides various native strategies for message redelivery and erro
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.
When you use Spring for Apache Pulsar, you can enable this property by setting the `spring.pulsar.reactive.consumer.ack-timeout` Boot property.
If this 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 also specify this property directly as a Pulsar consumer property via a <<reactive-consumer-customizer,consumer customizer>> such as:
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]
----
@@ -404,10 +393,9 @@ ReactiveMessageConsumerBuilderCustomizer<String> consumerCustomizer() {
----
==== 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 by setting `spring.pulsar.reactive.consumer.negative-ack-redelivery-delay`.
You can also set it directly as a Pulsar consumer property via a <<reactive-consumer-customizer,consumer customizer>> such as:
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]
----
@@ -476,7 +464,7 @@ The easy way to solve this is to provide a DLQ topic name always.
=== 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.reactive.reader.*`] application properties.
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