diff --git a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc b/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc index 1b9fbe8a..c3dd5d1f 100644 --- a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc @@ -860,7 +860,7 @@ If you want to specify some advanced backoff options for ack timeout with differ ---- @EnablePulsar @Configuration -static class AckTimeoutRedeliveryConfig { +class AckTimeoutRedeliveryConfig { @PulsarListener(subscriptionName = "withAckTimeoutRedeliveryBackoffSubscription", topics = "withAckTimeoutRedeliveryBackoff-test-topic", @@ -871,7 +871,7 @@ static class AckTimeoutRedeliveryConfig { } @Bean - public RedeliveryBackoff ackTimeoutRedeliveryBackoff() { + RedeliveryBackoff ackTimeoutRedeliveryBackoff() { return MultiplierRedeliveryBackoff.builder().minDelayMs(1000).maxDelayMs(10 * 1000).multiplier(2) .build(); } @@ -909,7 +909,7 @@ Here is an example: ---- @EnablePulsar @Configuration -static class NegativeAckRedeliveryConfig { +class NegativeAckRedeliveryConfig { @PulsarListener(subscriptionName = "withNegRedeliveryBackoffSubscription", topics = "withNegRedeliveryBackoff-test-topic", negativeAckRedeliveryBackoff = "redeliveryBackoff", @@ -919,7 +919,7 @@ static class NegativeAckRedeliveryConfig { } @Bean - public RedeliveryBackoff redeliveryBackoff() { + RedeliveryBackoff redeliveryBackoff() { return MultiplierRedeliveryBackoff.builder().minDelayMs(1000).maxDelayMs(10 * 1000).multiplier(2) .build(); } @@ -940,7 +940,7 @@ Let us see some details around this feature in action by inspecting some code sn ---- @EnablePulsar @Configuration -static class DeadLetterPolicyConfig { +class DeadLetterPolicyConfig { @PulsarListener(id = "deadLetterPolicyListener", subscriptionName = "deadLetterPolicySubscription", topics = "topic-with-dlp", deadLetterPolicy = "deadLetterPolicy", @@ -997,10 +997,10 @@ Let us see some details by examining a few code snippets. ---- @EnablePulsar @Configuration -static class PulsarConsumerErrorHandlerConfig { +class PulsarConsumerErrorHandlerConfig { - @Bean - public PulsarConsumerErrorHandler pulsarConsumerErrorHandler( + @Bean + PulsarConsumerErrorHandler pulsarConsumerErrorHandler( PulsarTemplate pulsarTemplate) { return new DefaultPulsarConsumerErrorHandler<>( new PulsarDeadLetterPublishingRecoverer<>(pulsarTemplate, (c, m) -> "my-foo-dlt"), new FixedBackOff(100, 10)); @@ -1083,18 +1083,18 @@ Towards this extent, you can use a `PulsarConsumerErrorHandler` with the followi [source, java] ---- @Bean -public PulsarConsumerErrorHandler pulsarConsumerErrorHandler(PulsarClient pulsarClient) { -PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, Map.of()); - PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); +PulsarConsumerErrorHandler pulsarConsumerErrorHandler(PulsarClient pulsarClient) { + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, Map.of()); + PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); - BiFunction, Message, String> destinationResolver = - (c, m) -> "my-foo-dlt"; + BiFunction, Message, String> destinationResolver = + (c, m) -> "my-foo-dlt"; - final PulsarDeadLetterPublishingRecoverer pulsarDeadLetterPublishingRecoverer = - new PulsarDeadLetterPublishingRecoverer<>(pulsarTemplate, destinationResolver); + PulsarDeadLetterPublishingRecoverer pulsarDeadLetterPublishingRecoverer = + new PulsarDeadLetterPublishingRecoverer<>(pulsarTemplate, destinationResolver); - return new DefaultPulsarConsumerErrorHandler<>(pulsarDeadLetterPublishingRecoverer, - new FixedBackOff(100, 5)); + return new DefaultPulsarConsumerErrorHandler<>(pulsarDeadLetterPublishingRecoverer, + new FixedBackOff(100, 5)); } ---- ==== @@ -1121,17 +1121,17 @@ First, let us look at a batch `PulsarListener` method. @PulsarListener(subscriptionName = "batch-demo-5-sub", topics = "batch-demo-4", batch = true, concurrency = "3", subscriptionType = SubscriptionType.Failover, pulsarConsumerErrorHandler = "pulsarConsumerErrorHandler", ackMode = AckMode.MANUAL) -public void listen(List> data, Consumer consumer, Acknowledgment acknowledgment) { +void listen(List> data, Consumer consumer, Acknowledgment acknowledgment) { for (Message datum : data) { - if (datum.getValue() == 5) { + if (datum.getValue() == 5) { throw new PulsarBatchListenerFailedException("failed", datum); - } - acknowledgement.acknowledge(datum.getMessageId()); - } + } + acknowledgement.acknowledge(datum.getMessageId()); + } } @Bean -public PulsarConsumerErrorHandler pulsarConsumerErrorHandler( +PulsarConsumerErrorHandler pulsarConsumerErrorHandler( PulsarTemplate pulsarTemplate) { return new DefaultPulsarConsumerErrorHandler<>( new PulsarDeadLetterPublishingRecoverer<>(pulsarTemplate, (c, m) -> "my-foo-dlt"), new FixedBackOff(100, 10));