From df00e9b67a3c15ce1636531c54ea20d9b0de85e3 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Mon, 12 Sep 2022 12:17:26 -0400 Subject: [PATCH] Enabling ackMode on PulsarListener --- .../autoconfigure/PulsarProperties.java | 8 ++-- .../pulsar/annotation/PulsarListener.java | 7 ++++ ...arListenerAnnotationBeanPostProcessor.java | 1 + ...bstractPulsarListenerContainerFactory.java | 10 +++-- .../AbstractPulsarListenerEndpoint.java | 11 ++++++ ...currentPulsarListenerContainerFactory.java | 3 -- .../pulsar/config/PulsarListenerEndpoint.java | 3 ++ .../config/PulsarListenerEndpointAdapter.java | 6 +++ .../pulsar/listener/AckMode.java | 39 +++++++++++++++++++ ...DefaultPulsarMessageListenerContainer.java | 25 +++++------- .../listener/PulsarContainerProperties.java | 20 ---------- .../core/ConsumerAcknowledgmentTests.java | 5 ++- ...efaultPulsarConsumerErrorHandlerTests.java | 10 ++--- .../pulsar/listener/PulsarListenerTests.java | 12 ++++++ 14 files changed, 108 insertions(+), 52 deletions(-) create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/listener/AckMode.java diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java index 774c79a3..58304eea 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java @@ -38,7 +38,7 @@ import org.apache.pulsar.common.schema.SchemaType; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.boot.context.properties.PropertyMapper; -import org.springframework.pulsar.listener.PulsarContainerProperties; +import org.springframework.pulsar.listener.AckMode; /** * Configuration properties for Spring for Apache Pulsar. @@ -1308,18 +1308,18 @@ public class PulsarProperties { /** * AckMode for acknowledgements. Allowed values are RECORD, BATCH, MANUAL. */ - private PulsarContainerProperties.AckMode ackMode; + private AckMode ackMode; /** * SchemaType of the consumed messages. */ private SchemaType schemaType; - public PulsarContainerProperties.AckMode getAckMode() { + public AckMode getAckMode() { return this.ackMode; } - public void setAckMode(PulsarContainerProperties.AckMode ackMode) { + public void setAckMode(AckMode ackMode) { this.ackMode = ackMode; } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java index 20e08989..29ac394b 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java @@ -27,6 +27,7 @@ import org.apache.pulsar.common.schema.SchemaType; import org.springframework.messaging.handler.annotation.MessageMapping; import org.springframework.pulsar.config.PulsarListenerContainerFactory; import org.springframework.pulsar.config.PulsarListenerEndpointRegistry; +import org.springframework.pulsar.listener.AckMode; /** * Annotation that marks a method to be the target of a Pulsar message listener on the @@ -179,4 +180,10 @@ public @interface PulsarListener { */ String deadLetterPolicy() default ""; + /** + * Override the container default ack mode of BATCH. + * @return ack mode used by the listener + */ + AckMode ackMode() default AckMode.BATCH; + } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java index 7220e981..531af750 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java @@ -349,6 +349,7 @@ public class PulsarListenerAnnotationBeanPostProcessor endpoint.setTopicPattern(topicPattern); endpoint.setSubscriptionType(getEndpointSubscriptionType(pulsarListener)); endpoint.setSchemaType(pulsarListener.schemaType()); + endpoint.setAckMode(pulsarListener.ackMode()); String concurrency = pulsarListener.concurrency(); if (StringUtils.hasText(concurrency)) { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java index e9c78be0..ad086fb7 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java @@ -28,6 +28,7 @@ import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.core.log.LogAccessor; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.listener.AbstractPulsarMessageListenerContainer; +import org.springframework.pulsar.listener.AckMode; import org.springframework.pulsar.listener.PulsarContainerProperties; import org.springframework.pulsar.support.JavaUtils; import org.springframework.pulsar.support.MessageConverter; @@ -148,16 +149,19 @@ public abstract class AbstractPulsarListenerContainerFactory private Integer concurrency; + private AckMode ackMode; + @Override public void setBeanFactory(BeanFactory beanFactory) throws BeansException { this.beanFactory = beanFactory; @@ -244,4 +247,12 @@ public abstract class AbstractPulsarListenerEndpoint this.concurrency = concurrency; } + public AckMode getAckMode() { + return this.ackMode; + } + + public void setAckMode(AckMode ackMode) { + this.ackMode = ackMode; + } + } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactory.java index e74dfd6e..ed5526f9 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactory.java @@ -48,7 +48,6 @@ public class ConcurrentPulsarListenerContainerFactory @Override protected ConcurrentPulsarMessageListenerContainer createContainerInstance(PulsarListenerEndpoint endpoint) { - PulsarContainerProperties properties = new PulsarContainerProperties(); Collection topics = endpoint.getTopics(); String topicPattern = endpoint.getTopicPattern(); @@ -57,13 +56,11 @@ public class ConcurrentPulsarListenerContainerFactory final String[] topics1 = topics.toArray(new String[0]); properties.setTopics(topics1); } - if (StringUtils.hasText(topicPattern)) { properties.setTopicsPattern(topicPattern); } final String subscriptionName = endpoint.getSubscriptionName(); - if (StringUtils.hasText(subscriptionName)) { properties.setSubscriptionName(endpoint.getSubscriptionName()); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpoint.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpoint.java index 4b1d470c..a4676300 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpoint.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpoint.java @@ -23,6 +23,7 @@ import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.common.schema.SchemaType; import org.springframework.lang.Nullable; +import org.springframework.pulsar.listener.AckMode; import org.springframework.pulsar.listener.PulsarMessageListenerContainer; import org.springframework.pulsar.support.MessageConverter; @@ -64,4 +65,6 @@ public interface PulsarListenerEndpoint { @Nullable Integer getConcurrency(); + AckMode getAckMode(); + } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointAdapter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointAdapter.java index 9413c0e1..3b427835 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointAdapter.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointAdapter.java @@ -24,6 +24,7 @@ import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.common.schema.SchemaType; import org.springframework.lang.Nullable; +import org.springframework.pulsar.listener.AckMode; import org.springframework.pulsar.listener.PulsarMessageListenerContainer; import org.springframework.pulsar.support.MessageConverter; @@ -92,4 +93,9 @@ public class PulsarListenerEndpointAdapter implements PulsarListenerEndpoint { return null; } + @Override + public AckMode getAckMode() { + return null; + } + } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AckMode.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AckMode.java new file mode 100644 index 00000000..c7613fb6 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AckMode.java @@ -0,0 +1,39 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.listener; + +/** + * Enumeration for ack mode. + * + * @author Soby Chacko + */ +public enum AckMode { + + /** + * Batch ack mode. + */ + BATCH, + /** + * Record ack mode. + */ + RECORD, + /** + * Manual ack mode. + */ + MANUAL; + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java index 1c41216d..0177db1d 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java @@ -286,14 +286,13 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess if (messageList != null && messageList.size() > 0) { if (this.batchMessageListener instanceof PulsarBatchAcknowledgingMessageListener) { this.batchMessageListener.received(this.consumer, messageList, - this.containerProperties - .getAckMode() == PulsarContainerProperties.AckMode.MANUAL - ? new ConsumerBatchAcknowledgment(this.consumer) : null); + this.containerProperties.getAckMode() == AckMode.MANUAL + ? new ConsumerBatchAcknowledgment(this.consumer) : null); } else { this.batchMessageListener.received(this.consumer, messageList); } - if (this.containerProperties.getAckMode() == PulsarContainerProperties.AckMode.BATCH) { + if (this.containerProperties.getAckMode() == AckMode.BATCH) { try { if (isSharedSubscriptionType()) { this.consumer.acknowledge(messages); @@ -332,15 +331,13 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess try { if (this.listener instanceof PulsarAcknowledgingMessageListener) { this.listener.received(this.consumer, message, - this.containerProperties - .getAckMode() == PulsarContainerProperties.AckMode.MANUAL - ? new ConsumerAcknowledgment(this.consumer, message) - : null); + this.containerProperties.getAckMode() == AckMode.MANUAL + ? new ConsumerAcknowledgment(this.consumer, message) : null); } else if (this.listener != null) { this.listener.received(this.consumer, message); } - if (this.containerProperties.getAckMode() == PulsarContainerProperties.AckMode.RECORD) { + if (this.containerProperties.getAckMode() == AckMode.RECORD) { handleAck(message); } if (inRetryMode.get()) { @@ -352,12 +349,10 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess invokeRecordListenerErrorHandler(inRetryMode, message, e); } else { - if (this.containerProperties - .getAckMode() == PulsarContainerProperties.AckMode.RECORD) { + if (this.containerProperties.getAckMode() == AckMode.RECORD) { this.consumer.negativeAcknowledge(message); } - else if (this.containerProperties - .getAckMode() == PulsarContainerProperties.AckMode.BATCH) { + else if (this.containerProperties.getAckMode() == AckMode.BATCH) { this.nackableMessages.add(message.getMessageId()); } } @@ -366,7 +361,7 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess while (inRetryMode.get()); } // All the records are processed at this point. Handle acks. - if (this.containerProperties.getAckMode() == PulsarContainerProperties.AckMode.BATCH) { + if (this.containerProperties.getAckMode() == AckMode.BATCH) { handleAcks(messages); } } @@ -449,7 +444,7 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess this.pulsarConsumerErrorHandler.recoverMessage(this.consumer, message, e); // retries exhausted - if record ackmode, acknowledge, otherwise normal // batch ack at the end - if (this.containerProperties.getAckMode() == PulsarContainerProperties.AckMode.RECORD) { + if (this.containerProperties.getAckMode() == AckMode.RECORD) { handleAck(message); } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java index 214fb690..fe91fcc1 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java @@ -38,26 +38,6 @@ public class PulsarContainerProperties { private Duration consumerStartTimeout = DEFAULT_CONSUMER_START_TIMEOUT; - /** - * Enumeration for ack mode. - */ - public enum AckMode { - - /** - * Batch ack mode. - */ - BATCH, - /** - * Record ack mode. - */ - RECORD, - /** - * Manual ack mode. - */ - MANUAL; - - } - private String[] topics; private String topicsPattern; diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java index 8df8b591..6ce95bde 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java @@ -46,6 +46,7 @@ import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.Schema; import org.junit.jupiter.api.Test; +import org.springframework.pulsar.listener.AckMode; import org.springframework.pulsar.listener.Acknowledgement; import org.springframework.pulsar.listener.DefaultPulsarMessageListenerContainer; import org.springframework.pulsar.listener.PulsarAcknowledgingMessageListener; @@ -73,7 +74,7 @@ class ConsumerAcknowledgmentTests extends AbstractContainerBaseTests { pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { }); pulsarContainerProperties.setSchema(Schema.STRING); - pulsarContainerProperties.setAckMode(PulsarContainerProperties.AckMode.RECORD); + pulsarContainerProperties.setAckMode(AckMode.RECORD); DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); container.start(); @@ -235,7 +236,7 @@ class ConsumerAcknowledgmentTests extends AbstractContainerBaseTests { pulsarContainerProperties.setMessageListener(pulsarAcknowledgingMessageListener); pulsarContainerProperties.setSchema(Schema.STRING); - pulsarContainerProperties.setAckMode(PulsarContainerProperties.AckMode.MANUAL); + pulsarContainerProperties.setAckMode(AckMode.MANUAL); DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); container.start(); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandlerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandlerTests.java index 3e9a4f82..71d272ff 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandlerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandlerTests.java @@ -246,7 +246,7 @@ public class DefaultPulsarConsumerErrorHandlerTests extends AbstractContainerBas pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener); pulsarContainerProperties.setSchema(Schema.INT32); - pulsarContainerProperties.setAckMode(PulsarContainerProperties.AckMode.MANUAL); + pulsarContainerProperties.setAckMode(AckMode.MANUAL); DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); PulsarTemplate mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS); @@ -316,7 +316,7 @@ public class DefaultPulsarConsumerErrorHandlerTests extends AbstractContainerBas pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener); pulsarContainerProperties.setSchema(Schema.INT32); - pulsarContainerProperties.setAckMode(PulsarContainerProperties.AckMode.MANUAL); + pulsarContainerProperties.setAckMode(AckMode.MANUAL); DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); PulsarTemplate mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS); @@ -385,7 +385,7 @@ public class DefaultPulsarConsumerErrorHandlerTests extends AbstractContainerBas pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener); pulsarContainerProperties.setSchema(Schema.INT32); - pulsarContainerProperties.setAckMode(PulsarContainerProperties.AckMode.MANUAL); + pulsarContainerProperties.setAckMode(AckMode.MANUAL); DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); PulsarTemplate mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS); @@ -460,7 +460,7 @@ public class DefaultPulsarConsumerErrorHandlerTests extends AbstractContainerBas pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener); pulsarContainerProperties.setSchema(Schema.INT32); - pulsarContainerProperties.setAckMode(PulsarContainerProperties.AckMode.MANUAL); + pulsarContainerProperties.setAckMode(AckMode.MANUAL); DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); PulsarTemplate mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS); @@ -531,7 +531,7 @@ public class DefaultPulsarConsumerErrorHandlerTests extends AbstractContainerBas pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener); pulsarContainerProperties.setSchema(Schema.INT32); - pulsarContainerProperties.setAckMode(PulsarContainerProperties.AckMode.MANUAL); + pulsarContainerProperties.setAckMode(AckMode.MANUAL); DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); PulsarTemplate mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java index 414fd1c0..37a0f4b7 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java @@ -206,6 +206,13 @@ public class PulsarListenerTests extends AbstractContainerBaseTests { assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue(); } + @Test + void ackModeAppliedToContainerFromListener(@Autowired PulsarListenerEndpointRegistry registry) { + final PulsarContainerProperties pulsarContainerProperties = registry.getListenerContainer("ackMode-test-id") + .getContainerProperties(); + assertThat(pulsarContainerProperties.getAckMode()).isEqualTo(AckMode.RECORD); + } + @EnablePulsar @Configuration static class TestPulsarListenersForBasicScenario { @@ -228,6 +235,11 @@ public class PulsarListenerTests extends AbstractContainerBaseTests { latch2.countDown(); } + @PulsarListener(id = "ackMode-test-id", subscriptionName = "ackModeTest-sub", topics = "ackModeTest-topic", + ackMode = AckMode.RECORD) + void ackModeTestListener(String message) { + } + } }