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

Use Duration, DataSize and boxed types in PulsarProperties (#191)

Browse Source 
See #119
This commit is contained in:
Christophe Bornet
2022-11-04 02:21:46 +01:00
committed by GitHub
parent f466ce8732
commit b4203bedc8
4 changed files with 275 additions and 271 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
spring-pulsar-docs/src/main/asciidoc/pulsar.adoc
View File

@@ -889,7 +889,7 @@ Apache Pulsar provides various native strategies for message redelivery and erro
===== Specifying Acknowledgment Timeout for Message Redelivery
By default, Pulsar consumers will not redeliver messages unless the consumer crashes, but you can change this behavior by setting an ack timeout on the Pulsar consumer.
When using Spring for Apache Pulsar, we can enable this property by setting the Boot property `spring.pulsar.consumer.ack-timeout-millis`.
When using Spring for Apache Pulsar, we can enable this property by setting the Boot property `spring.pulsar.consumer.ack-timeout`.
If this property has a value above zero, then if Pulsar consumer does not acknowledge a message within that timeout period, then the message will be redelivered.
You can also specify this property directly as a Pulsar consumer property on the `PulsarListener` itself as shown below:
@@ -898,14 +898,14 @@ You can also specify this property directly as a Pulsar consumer property on the
[source, java]
----
@PulsarListener(subscriptionName = "subscription-1", topics = "topic-1"
properties = {"ackTimeoutMillis=60000"})
properties = {"ackTimeout=60s"})
public void listen(String s) {
...
}
----
====
When specifying `ackTimeoutMillis` as seen in the above `PulsarListener` method, then if the consumer does not send an acknowledgement within 60 seconds, the message will be redelivered by Pulsar to the consumer.
When specifying `ackTimeout` as seen in the above `PulsarListener` method, then if the consumer does not send an acknowledgement within 60 seconds, the message will be redelivered by Pulsar to the consumer.
If you want to specify some advanced backoff options for ack timeout with different delays, then you can do the following:
@@ -919,7 +919,7 @@ class AckTimeoutRedeliveryConfig {
@PulsarListener(subscriptionName = "withAckTimeoutRedeliveryBackoffSubscription",
topics = "withAckTimeoutRedeliveryBackoff-test-topic",
ackTimeoutRedeliveryBackoff = "ackTimeoutRedeliveryBackoff",
properties = { "ackTimeoutMillis=60000" })
properties = { "ackTimeout=60s" })
void listen(String msg) {
// some long-running process that may cause an ack timeout
}
@@ -941,14 +941,14 @@ We provide the backoff bean to the `PulsarListener` annotation by setting the `a
===== Specifying Negative Acknowledgment Redelivery
When acknowledging negatively, Pulsar consumer allows you to specify how the application want the message to be re-delivered.
The default is to redeliver the message in 1 minute, but you can change it by providing `spring.pulsar.consumer.negative-ack-redelivery-delay-micros`.
The default is to redeliver the message in 1 minute, but you can change it by providing `spring.pulsar.consumer.negative-ack-redelivery-delay`.
You can also set it as a consumer property directly on `PulsarListener` as shown below:
====
[source, java]
----
@PulsarListener(subscriptionName = "subscription-1", topics = "topic-1"
properties = {"negativeAckRedeliveryDelayMicros=10000"})
properties = {"negativeAckRedeliveryDelay=10ms"})
public void listen(String s) {
...
}
@@ -998,7 +998,7 @@ class DeadLetterPolicyConfig {
@PulsarListener(id = "deadLetterPolicyListener", subscriptionName = "deadLetterPolicySubscription",
topics = "topic-with-dlp", deadLetterPolicy = "deadLetterPolicy",
subscriptionType = SubscriptionType.Shared, properties = { "ackTimeoutMillis=1" })
subscriptionType = SubscriptionType.Shared, properties = { "ackTimeout=1s" })
void listen(String msg) {
throw new RuntimeException("fail " + msg);
}
@@ -1023,8 +1023,8 @@ This bean specifies a number of things, such as the max delivery - 10 in this ca
If you don't specify a DLQ topic name, then it defaults to `<topicname>-<subscriptionname>-DLQ` in Pulsar.
Next, we provide this bean name to `PulsarListener` using the property `deadLetterPolicy`.
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 `ackTimeoutMillis` value of 1 millisecond.
The idea is that the code throws the exception and if Pulsar does not receive an ack within 1 millisecond, it does a retry.
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 for 10 times, (as that is our max redelivery count in the `DeadLetterPolicy`), then Pulsar consumer publishes the messages to the DQL topic.
We have another `PulsarListener` that is listening on the DLQ topic to receive data as it is published to the DLQ topic.

8
spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java
View File

@@ -16,6 +16,8 @@
package org.springframework.pulsar.autoconfigure;
import java.time.Duration;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
@@ -28,6 +30,7 @@ import org.springframework.pulsar.config.PulsarListenerBeanNames;
import org.springframework.pulsar.core.PulsarConsumerFactory;
import org.springframework.pulsar.listener.PulsarContainerProperties;
import org.springframework.pulsar.observation.PulsarListenerObservationConvention;
import org.springframework.util.unit.DataSize;
import io.micrometer.observation.ObservationRegistry;
@@ -62,8 +65,9 @@ public class PulsarAnnotationDrivenConfiguration {
PulsarProperties.Listener listenerProperties = this.pulsarProperties.getListener();
map.from(listenerProperties::getSchemaType).to(containerProperties::setSchemaType);
map.from(listenerProperties::getAckMode).to(containerProperties::setAckMode);
map.from(listenerProperties::getBatchTimeoutMillis).to(containerProperties::setBatchTimeoutMillis);
map.from(listenerProperties::getMaxNumBytes).to(containerProperties::setMaxNumBytes);
map.from(listenerProperties::getBatchTimeout).asInt(Duration::toMillis)
.to(containerProperties::setBatchTimeoutMillis);
map.from(listenerProperties::getMaxNumBytes).asInt(DataSize::toBytes).to(containerProperties::setMaxNumBytes);
map.from(listenerProperties::getMaxNumMessages).to(containerProperties::setMaxNumMessages);
return new ConcurrentPulsarListenerContainerFactory<>(consumerFactoryProvider.getIfAvailable(),

518
spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java
View File

File diff suppressed because it is too large Load Diff

2
spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java
View File

@@ -210,7 +210,7 @@ class PulsarAutoConfigurationTests {
void consumerBatchPropertiesAreHonored() {
contextRunner
.withPropertyValues("spring.pulsar.listener.max-num-messages=10",
"spring.pulsar.listener.max-num-bytes=101", "spring.pulsar.listener.batch-timeout-millis=50")
"spring.pulsar.listener.max-num-bytes=101B", "spring.pulsar.listener.batch-timeout=50ms")
.run((context -> assertThat(context).hasNotFailed()
.getBean(ConcurrentPulsarListenerContainerFactory.class).extracting("containerProperties")
.hasFieldOrPropertyWithValue("maxNumMessages", 10)