diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/message-consumption.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/message-consumption.adoc index dc226a7c..5758c3cc 100644 --- a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/message-consumption.adoc +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/message-consumption.adoc @@ -625,12 +625,12 @@ Apache Pulsar provides various native strategies for message redelivery and erro 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. -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`: +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 `ackTimeoutMillis` property in the `properties` attribute of `@PulsarListener`: [source, java] ---- @PulsarListener(subscriptionName = "subscription-1", topics = "topic-1" - properties = {"ackTimeout=60s"}) + properties = {"ackTimeoutMillis=60000"}) public void listen(String s) { ... } @@ -649,7 +649,7 @@ class AckTimeoutRedeliveryConfig { @PulsarListener(subscriptionName = "withAckTimeoutRedeliveryBackoffSubscription", topics = "withAckTimeoutRedeliveryBackoff-test-topic", ackTimeoutRedeliveryBackoff = "ackTimeoutRedeliveryBackoff", - properties = { "ackTimeout=60s" }) + properties = { "ackTimeoutMillis=60000" }) void listen(String msg) { // some long-running process that may cause an ack timeout } @@ -720,7 +720,7 @@ class DeadLetterPolicyConfig { @PulsarListener(id = "deadLetterPolicyListener", subscriptionName = "deadLetterPolicySubscription", topics = "topic-with-dlp", deadLetterPolicy = "deadLetterPolicy", - subscriptionType = SubscriptionType.Shared, properties = { "ackTimeout=1s" }) + subscriptionType = SubscriptionType.Shared, properties = { "ackTimeoutMillis=1000" }) void listen(String msg) { throw new RuntimeException("fail " + msg); } @@ -743,7 +743,7 @@ This bean specifies a number of things, such as the max delivery (10, in this ca If you do not specify a DLQ topic name, it defaults to `--DLQ` in Pulsar. Next, we provide this bean name to `PulsarListener` by setting the `deadLetterPolicy` property. Note that the `PulsarListener` 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. +This code is primarily for demonstration purposes, so we provide an `ackTimeoutMillis` value of 1000. 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 `PulsarListener` that listens on the DLQ topic to receive data as it is published to the DLQ topic. diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar/reactive-message-consumption.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar/reactive-message-consumption.adoc index 1209c3a3..e2489eff 100644 --- a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar/reactive-message-consumption.adoc +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar/reactive-message-consumption.adoc @@ -88,6 +88,7 @@ Flux> listen(Flux 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`. @@ -293,7 +294,7 @@ You can specify this property directly as a Pulsar consumer property via a < consumerCustomizer() { - return b -> b.property("ackTimeout", "60s"); + return b -> b.property("ackTimeoutMillis", "60000"); } ---- @@ -342,7 +343,7 @@ class DeadLetterPolicyConfig { @Bean ReactiveMessageConsumerBuilderCustomizer ackTimeoutCustomizer() { - return b -> b.property("ackTimeout", "1s"); + return b -> b.property("ackTimeoutMillis", "1000"); } } ---- @@ -352,7 +353,7 @@ This bean specifies a number of things, such as the max delivery (10, in this ca If you do not specify a DLQ topic name, it defaults to `--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. +This code is primarily for demonstration purposes, so we provide an `ackTimeoutMillis` value of 1000. 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.