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 93709d68..bab0433d 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 @@ -157,9 +157,17 @@ public @interface PulsarListener { * be a property placeholder or SpEL expression that evaluates to a {@link Number}, in * which case {@link Number#intValue()} is used to obtain the value. *

- * SpEL {@code #{...}} and property place holders {@code ${...}} are supported. + * SpEL {@code #{...}} and property placeholders {@code ${...}} are supported. * @return the concurrency. */ String concurrency() default ""; + /** + * The bean name or a 'SpEL' expression that resolves to a + * {@link org.apache.pulsar.client.api.RedeliveryBackoff} to use on the consumer to + * control the redelivery backoff of messages after a negative ack. + * @return the bean name or empty string to not set the backoff + */ + String negativeAckRedeliveryBackoff() default ""; + } 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 c10279e6..9d6030ed 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 @@ -39,6 +39,7 @@ import java.util.function.BiFunction; import java.util.stream.Collectors; import org.apache.commons.logging.LogFactory; +import org.apache.pulsar.client.api.RedeliveryBackoff; import org.apache.pulsar.client.api.SubscriptionType; import org.springframework.aop.framework.Advised; @@ -360,6 +361,24 @@ public class PulsarListenerAnnotationBeanPostProcessor resolvePulsarProperties(endpoint, pulsarListener.properties()); endpoint.setBatchListener(pulsarListener.batch()); endpoint.setBeanFactory(this.beanFactory); + + resolveNegativeAckRedeliveryBackoff(endpoint, pulsarListener); + } + + private void resolveNegativeAckRedeliveryBackoff(MethodPulsarListenerEndpoint endpoint, + PulsarListener pulsarListener) { + Object negativeAckRedeliveryBackoff = resolveExpression(pulsarListener.negativeAckRedeliveryBackoff()); + if (negativeAckRedeliveryBackoff instanceof RedeliveryBackoff) { + endpoint.setNegativeAckRedeliveryBackoff((RedeliveryBackoff) negativeAckRedeliveryBackoff); + } + else { + String negativeAckRedeliveryBackoffBeanName = resolveExpressionAsString( + pulsarListener.negativeAckRedeliveryBackoff(), "negativeAckRedeliveryBackoff"); + if (StringUtils.hasText(negativeAckRedeliveryBackoffBeanName)) { + endpoint.setNegativeAckRedeliveryBackoff( + this.beanFactory.getBean(negativeAckRedeliveryBackoffBeanName, RedeliveryBackoff.class)); + } + } } private Integer resolveExpressionAsInteger(String value, String attribute) { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java index 0799b68e..63ca9311 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java @@ -26,6 +26,7 @@ import org.apache.commons.logging.LogFactory; import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.Messages; +import org.apache.pulsar.client.api.RedeliveryBackoff; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.impl.schema.AvroSchema; import org.apache.pulsar.client.impl.schema.JSONSchema; @@ -74,6 +75,8 @@ public class MethodPulsarListenerEndpoint extends AbstractPulsarListenerEndpo private SmartMessageConverter messagingConverter; + private RedeliveryBackoff negativeAckRedeliveryBackoff; + public void setBean(Object bean) { this.bean = bean; } @@ -172,6 +175,8 @@ public class MethodPulsarListenerEndpoint extends AbstractPulsarListenerEndpo final SchemaType type = pulsarContainerProperties.getSchema().getSchemaInfo().getType(); pulsarContainerProperties.setSchemaType(type); + container.setNegativeAckRedeliveryBackoff(this.negativeAckRedeliveryBackoff); + return messageListener; } @@ -245,4 +250,8 @@ public class MethodPulsarListenerEndpoint extends AbstractPulsarListenerEndpo this.messagingConverter = messagingConverter; } + public void setNegativeAckRedeliveryBackoff(RedeliveryBackoff negativeAckRedeliveryBackoff) { + this.negativeAckRedeliveryBackoff = negativeAckRedeliveryBackoff; + } + } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java index 6688c5f5..2dbce48d 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java @@ -26,6 +26,7 @@ import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.ConsumerBuilder; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.api.RedeliveryBackoff; import org.apache.pulsar.client.api.Schema; import org.springframework.util.CollectionUtils; @@ -80,6 +81,12 @@ public class DefaultPulsarConsumerFactory implements PulsarConsumerFactory consumerBuilder.loadConf(properties); } + if (properties.containsKey("negativeAckRedeliveryBackoff")) { + final RedeliveryBackoff negativeAckRedeliveryBackoff = (RedeliveryBackoff) properties + .get("negativeAckRedeliveryBackoff"); + consumerBuilder.negativeAckRedeliveryBackoff(negativeAckRedeliveryBackoff); + } + consumerBuilder.batchReceivePolicy(batchReceivePolicy); Consumer consumer = consumerBuilder.subscribe(); this.consumers.add(consumer); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java index 1d3d0250..3c13d5d0 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java @@ -17,6 +17,7 @@ package org.springframework.pulsar.listener; import org.apache.commons.logging.LogFactory; +import org.apache.pulsar.client.api.RedeliveryBackoff; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanNameAware; @@ -58,6 +59,8 @@ public abstract class AbstractPulsarMessageListenerContainer implements Pulsa private volatile boolean running = false; + protected RedeliveryBackoff negativeAckRedeliveryBackoff; + @SuppressWarnings("unchecked") protected AbstractPulsarMessageListenerContainer(PulsarConsumerFactory pulsarConsumerFactory, PulsarContainerProperties pulsarContainerProperties) { @@ -173,4 +176,13 @@ public abstract class AbstractPulsarMessageListenerContainer implements Pulsa } } + @Override + public void setNegativeAckRedeliveryBackoff(RedeliveryBackoff redeliveryBackoff) { + this.negativeAckRedeliveryBackoff = redeliveryBackoff; + } + + public RedeliveryBackoff getNegativeAckRedeliveryBackoff() { + return this.negativeAckRedeliveryBackoff; + } + } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainer.java index 16244b17..60c22d3a 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainer.java @@ -108,6 +108,7 @@ public class ConcurrentPulsarMessageListenerContainer extends AbstractPulsarM this.executors.add(exec); container.getContainerProperties().setConsumerTaskExecutor(exec); } + container.setNegativeAckRedeliveryBackoff(this.negativeAckRedeliveryBackoff); } @Override @@ -126,4 +127,8 @@ public class ConcurrentPulsarMessageListenerContainer extends AbstractPulsarM return false; } + public List> getContainers() { + return this.containers; + } + } 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 91653790..429099e7 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 @@ -37,6 +37,7 @@ import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.MessageListener; import org.apache.pulsar.client.api.Messages; import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.api.RedeliveryBackoff; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.SubscriptionType; @@ -179,14 +180,15 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess } try { final PulsarContainerProperties pulsarContainerProperties = getPulsarContainerProperties(); - Map propertiesToOverride = extractPropertiesToOverride(pulsarContainerProperties); + Map propertiesToConsumer = extractDirectConsumerProperties(); + populateAllNecessaryPropertiesIfNeedBe(propertiesToConsumer); final BatchReceivePolicy batchReceivePolicy = new BatchReceivePolicy.Builder() .maxNumMessages(pulsarContainerProperties.getMaxNumMessages()) .maxNumBytes(pulsarContainerProperties.getMaxNumBytes()) .timeout(pulsarContainerProperties.getBatchTimeout(), TimeUnit.MILLISECONDS).build(); this.consumer = getPulsarConsumerFactory().createConsumer( - (Schema) pulsarContainerProperties.getSchema(), batchReceivePolicy, propertiesToOverride); + (Schema) pulsarContainerProperties.getSchema(), batchReceivePolicy, propertiesToConsumer); Assert.state(this.consumer != null, "Unable to create a consumer"); } catch (PulsarClientException e) { @@ -194,50 +196,49 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess } } - private Map extractPropertiesToOverride(PulsarContainerProperties pulsarContainerProperties) { - + private Map extractDirectConsumerProperties() { Properties propertyOverrides = this.containerProperties.getPulsarConsumerProperties(); + return propertyOverrides.entrySet().stream().collect(Collectors.toMap(e -> String.valueOf(e.getKey()), + Map.Entry::getValue, (prev, next) -> next, HashMap::new)); + } - final Map propOverridesAsMap = propertyOverrides.entrySet().stream().collect(Collectors - .toMap(e -> String.valueOf(e.getKey()), Map.Entry::getValue, (prev, next) -> next, HashMap::new)); - - final Map propertiesToOverride = new HashMap<>(propOverridesAsMap); - if (propertiesToOverride.containsKey("topicNames")) { - final String topicsFromMap = (String) propertiesToOverride.get("topicNames"); - final String[] topicNames = topicsFromMap.split(","); - final Set propertiesDefinedTopics = new HashSet<>(Arrays.stream(topicNames).toList()); + private void populateAllNecessaryPropertiesIfNeedBe(Map currentProperties) { + if (currentProperties.containsKey("topicNames")) { + final String topicsFromMap = (String) currentProperties.get("topicNames"); + final String[] topicNames = StringUtils.delimitedListToStringArray(topicsFromMap, ","); + final Set propertiesDefinedTopics = Set.of(topicNames); if (!propertiesDefinedTopics.isEmpty()) { - propertiesToOverride.put("topicNames", propertiesDefinedTopics); + currentProperties.put("topicNames", propertiesDefinedTopics); } } - - if (!propertiesToOverride.containsKey("subscriptionType")) { - final SubscriptionType subscriptionType = pulsarContainerProperties.getSubscriptionType(); + if (!currentProperties.containsKey("subscriptionType")) { + final SubscriptionType subscriptionType = this.containerProperties.getSubscriptionType(); if (subscriptionType != null) { - propertiesToOverride.put("subscriptionType", subscriptionType); + currentProperties.put("subscriptionType", subscriptionType); } } - if (!propertiesToOverride.containsKey("topicNames")) { - final String[] topics = pulsarContainerProperties.getTopics(); + if (!currentProperties.containsKey("topicNames")) { + final String[] topics = this.containerProperties.getTopics(); final Set listenerDefinedTopics = new HashSet<>(Arrays.stream(topics).toList()); if (!listenerDefinedTopics.isEmpty()) { - propertiesToOverride.put("topicNames", listenerDefinedTopics); + currentProperties.put("topicNames", listenerDefinedTopics); } } - - if (!propertiesToOverride.containsKey("topicsPattern")) { - final String topicsPattern = pulsarContainerProperties.getTopicsPattern(); + if (!currentProperties.containsKey("topicsPattern")) { + final String topicsPattern = this.containerProperties.getTopicsPattern(); if (topicsPattern != null) { - propertiesToOverride.put("topicsPattern", topicsPattern); + currentProperties.put("topicsPattern", topicsPattern); } } - - if (!propertiesToOverride.containsKey("subscriptionName")) { - if (StringUtils.hasText(pulsarContainerProperties.getSubscriptionName())) { - propertiesToOverride.put("subscriptionName", pulsarContainerProperties.getSubscriptionName()); + if (!currentProperties.containsKey("subscriptionName")) { + if (StringUtils.hasText(this.containerProperties.getSubscriptionName())) { + currentProperties.put("subscriptionName", this.containerProperties.getSubscriptionName()); } } - return propertiesToOverride; + final RedeliveryBackoff negativeAckRedeliveryBackoff = DefaultPulsarMessageListenerContainer.this.negativeAckRedeliveryBackoff; + if (negativeAckRedeliveryBackoff != null) { + currentProperties.put("negativeAckRedeliveryBackoff", negativeAckRedeliveryBackoff); + } } @Override @@ -253,7 +254,6 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess publishConsumerStartedEvent(); while (isRunning()) { Messages messages = null; - // Always receive messages in batch mode. try { messages = this.consumer.batchReceive(); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageListenerContainer.java index c386eb17..23da3b1b 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageListenerContainer.java @@ -16,6 +16,8 @@ package org.springframework.pulsar.listener; +import org.apache.pulsar.client.api.RedeliveryBackoff; + import org.springframework.beans.factory.DisposableBean; import org.springframework.context.SmartLifecycle; @@ -42,4 +44,6 @@ public interface PulsarMessageListenerContainer extends SmartLifecycle, Disposab throw new UnsupportedOperationException("This container doesn't support retrieving its properties"); } + void setNegativeAckRedeliveryBackoff(RedeliveryBackoff redeliveryBackoff); + } 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 ddfa78ce..e6c80291 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 @@ -24,7 +24,6 @@ import static org.mockito.Mockito.atMost; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; -import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -47,14 +46,12 @@ import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.Schema; import org.junit.jupiter.api.Test; -import org.springframework.beans.DirectFieldAccessor; import org.springframework.pulsar.listener.Acknowledgement; import org.springframework.pulsar.listener.DefaultPulsarMessageListenerContainer; import org.springframework.pulsar.listener.PulsarAcknowledgingMessageListener; import org.springframework.pulsar.listener.PulsarBatchMessageListener; import org.springframework.pulsar.listener.PulsarContainerProperties; import org.springframework.pulsar.listener.PulsarRecordMessageListener; -import org.springframework.util.Assert; /** * @author Soby Chacko @@ -80,7 +77,7 @@ class ConsumerAcknowledgmentTests extends AbstractContainerBaseTests { DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); container.start(); - final Consumer containerConsumer = spyOnConsumer(container); + final Consumer containerConsumer = ConsumerTestUtils.spyOnConsumer(container); CountDownLatch latch = new CountDownLatch(10); @@ -121,7 +118,7 @@ class ConsumerAcknowledgmentTests extends AbstractContainerBaseTests { DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); container.start(); - final Consumer containerConsumer = spyOnConsumer(container); + final Consumer containerConsumer = ConsumerTestUtils.spyOnConsumer(container); Map prodConfig = new HashMap<>(); prodConfig.put("topicName", "cons-ack-tests-012"); @@ -167,7 +164,7 @@ class ConsumerAcknowledgmentTests extends AbstractContainerBaseTests { DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); container.start(); - final Consumer containerConsumer = spyOnConsumer(container); + final Consumer containerConsumer = ConsumerTestUtils.spyOnConsumer(container); AtomicInteger ackCallCount = new AtomicInteger(0); doAnswer(invocation -> { @@ -242,7 +239,7 @@ class ConsumerAcknowledgmentTests extends AbstractContainerBaseTests { DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); container.start(); - final Consumer containerConsumer = spyOnConsumer(container); + final Consumer containerConsumer = ConsumerTestUtils.spyOnConsumer(container); CountDownLatch latch = new CountDownLatch(10); @@ -299,7 +296,7 @@ class ConsumerAcknowledgmentTests extends AbstractContainerBaseTests { DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); container.start(); - final Consumer containerConsumer = spyOnConsumer(container); + final Consumer containerConsumer = ConsumerTestUtils.spyOnConsumer(container); Map prodConfig = new HashMap<>(); prodConfig.put("topicName", "cons-ack-tests-015"); @@ -347,7 +344,7 @@ class ConsumerAcknowledgmentTests extends AbstractContainerBaseTests { DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); container.start(); - final Consumer containerConsumer = spyOnConsumer(container); + final Consumer containerConsumer = ConsumerTestUtils.spyOnConsumer(container); Map prodConfig = new HashMap<>(); prodConfig.put("topicName", "cons-ack-tests-016"); @@ -367,47 +364,4 @@ class ConsumerAcknowledgmentTests extends AbstractContainerBaseTests { pulsarClient.close(); } - private Consumer spyOnConsumer(DefaultPulsarMessageListenerContainer container) { - Consumer consumer = getPropertyValue(container, "listenerConsumer.consumer", Consumer.class); - consumer = spy(consumer); - new DirectFieldAccessor(getPropertyValue(container, "listenerConsumer")).setPropertyValue("consumer", consumer); - return consumer; - } - - /** - * Uses nested {@link DirectFieldAccessor}s to obtain a property using dotted notation - * to traverse fields; e.g. "foo.bar.baz" will obtain a reference to the baz field of - * the bar field of foo. Adopted from Spring Integration. - * @param root The object. - * @param propertyPath The path. - * @return The field. - */ - public static Object getPropertyValue(Object root, String propertyPath) { - Object value = null; - DirectFieldAccessor accessor = new DirectFieldAccessor(root); - String[] tokens = propertyPath.split("\\."); - for (int i = 0; i < tokens.length; i++) { - value = accessor.getPropertyValue(tokens[i]); - if (value != null) { - accessor = new DirectFieldAccessor(value); - } - else if (i == tokens.length - 1) { - return null; - } - else { - throw new IllegalArgumentException("intermediate property '" + tokens[i] + "' is null"); - } - } - return value; - } - - @SuppressWarnings("unchecked") - public static T getPropertyValue(Object root, String propertyPath, Class type) { - Object value = getPropertyValue(root, propertyPath); - if (value != null) { - Assert.isAssignable(type, value.getClass()); - } - return (T) value; - } - } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerTestUtils.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerTestUtils.java new file mode 100644 index 00000000..610e5fb0 --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerTestUtils.java @@ -0,0 +1,81 @@ +/* + * 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.core; + +import static org.mockito.Mockito.spy; + +import org.apache.pulsar.client.api.Consumer; + +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.pulsar.listener.DefaultPulsarMessageListenerContainer; +import org.springframework.util.Assert; + +public final class ConsumerTestUtils { + + private ConsumerTestUtils() { + + } + + /** + * Provides a Mockito spy object for the message listener container. + * @param container container to spy on + * @return the spied container object + */ + public static Consumer spyOnConsumer(DefaultPulsarMessageListenerContainer container) { + Consumer consumer = getPropertyValue(container, "listenerConsumer.consumer", Consumer.class); + consumer = spy(consumer); + new DirectFieldAccessor(getPropertyValue(container, "listenerConsumer")).setPropertyValue("consumer", consumer); + return consumer; + } + + /** + * Uses nested {@link DirectFieldAccessor}s to obtain a property using dotted notation + * to traverse fields; e.g. "foo.bar.baz" will obtain a reference to the baz field of + * the bar field of foo. Adopted from Spring Integration. + * @param root The object. + * @param propertyPath The path. + * @return The field. + */ + public static Object getPropertyValue(Object root, String propertyPath) { + Object value = null; + DirectFieldAccessor accessor = new DirectFieldAccessor(root); + String[] tokens = propertyPath.split("\\."); + for (int i = 0; i < tokens.length; i++) { + value = accessor.getPropertyValue(tokens[i]); + if (value != null) { + accessor = new DirectFieldAccessor(value); + } + else if (i == tokens.length - 1) { + return null; + } + else { + throw new IllegalArgumentException("intermediate property '" + tokens[i] + "' is null"); + } + } + return value; + } + + @SuppressWarnings("unchecked") + public static T getPropertyValue(Object root, String propertyPath, Class type) { + Object value = getPropertyValue(root, propertyPath); + if (value != null) { + Assert.isAssignable(type, value.getClass()); + } + return (T) value; + } + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java index 563332f0..f45376e7 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java @@ -16,6 +16,7 @@ package org.springframework.pulsar.listener; +import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.awaitility.Awaitility.await; import static org.mockito.ArgumentMatchers.any; @@ -30,8 +31,10 @@ import java.util.Map; import org.apache.pulsar.client.api.BatchReceivePolicy; import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.Messages; +import org.apache.pulsar.client.api.RedeliveryBackoff; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.client.impl.MultiplierRedeliveryBackoff; import org.junit.jupiter.api.Test; import org.springframework.pulsar.core.PulsarConsumerFactory; @@ -41,6 +44,35 @@ import org.springframework.pulsar.core.PulsarConsumerFactory; */ public class ConcurrentPulsarMessageListenerContainerTests { + @Test + @SuppressWarnings("unchecked") + void nackRedeliveryBackoffAppliedOnChildContainer() throws Exception { + PulsarConsumerFactory pulsarConsumerFactory = mock(PulsarConsumerFactory.class); + Consumer consumer = mock(Consumer.class); + + when(pulsarConsumerFactory.createConsumer(any(Schema.class), any(BatchReceivePolicy.class), any(Map.class))) + .thenReturn(consumer); + + when(consumer.batchReceive()).thenReturn(mock(Messages.class)); + + PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); + pulsarContainerProperties.setSchema(Schema.STRING); + pulsarContainerProperties.setSubscriptionType(SubscriptionType.Shared); + pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener) (cons, msg) -> { + }); + + ConcurrentPulsarMessageListenerContainer concurrentContainer = new ConcurrentPulsarMessageListenerContainer<>( + pulsarConsumerFactory, pulsarContainerProperties); + RedeliveryBackoff redeliveryBackoff = MultiplierRedeliveryBackoff.builder().minDelayMs(1000) + .maxDelayMs(5 * 1000).build(); + concurrentContainer.setNegativeAckRedeliveryBackoff(redeliveryBackoff); + + concurrentContainer.start(); + + final DefaultPulsarMessageListenerContainer childContainer = concurrentContainer.getContainers().get(0); + assertThat(childContainer.getNegativeAckRedeliveryBackoff()).isEqualTo(redeliveryBackoff); + } + @Test @SuppressWarnings("unchecked") void basicConcurrencyTesting() throws Exception { diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java index 2a00b07c..d74fc870 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java @@ -17,8 +17,14 @@ package org.springframework.pulsar.listener; import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import java.time.Duration; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.List; @@ -26,12 +32,18 @@ import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.RedeliveryBackoff; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.SubscriptionInitialPosition; +import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.client.impl.MultiplierRedeliveryBackoff; import org.junit.jupiter.api.Test; import org.springframework.pulsar.core.AbstractContainerBaseTests; +import org.springframework.pulsar.core.ConsumerTestUtils; import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; import org.springframework.pulsar.core.DefaultPulsarProducerFactory; import org.springframework.pulsar.core.PulsarTemplate; @@ -140,4 +152,54 @@ class DefaultPulsarMessageListenerContainerTests extends AbstractContainerBaseTe pulsarClient.close(); } + @Test + void negativeAckRedeliveryBackoff() throws Exception { + Map config = new HashMap<>(); + config.put("topicNames", Collections.singleton("dpmlct-015")); + config.put("subscriptionName", "dpmlct-sb-015"); + + RedeliveryBackoff redeliveryBackoff = MultiplierRedeliveryBackoff.builder().minDelayMs(1000) + .maxDelayMs(5 * 1000).build(); + config.put("negativeAckRedeliveryBackoff", redeliveryBackoff); + + final PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(getPulsarBrokerUrl()).build(); + final DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>( + pulsarClient, config); + CountDownLatch latch = new CountDownLatch(10); + PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); + pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { + latch.countDown(); + if (((String) msg.getValue()).endsWith("4")) { + throw new RuntimeException("fail"); + } + }); + pulsarContainerProperties.setSchema(Schema.STRING); + pulsarContainerProperties.setSubscriptionType(SubscriptionType.Shared); + DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( + pulsarConsumerFactory, pulsarContainerProperties); + container.start(); + + final Consumer containerConsumer = ConsumerTestUtils.spyOnConsumer(container); + + Map prodConfig = Collections.singletonMap("topicName", "dpmlct-015"); + final DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>( + pulsarClient, prodConfig); + final PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); + for (int i = 0; i < 5; i++) { + pulsarTemplate.send("hello john doe" + i); + } + assertThat(latch.await(30, TimeUnit.SECONDS)).isTrue(); + + // At this point, we should have 6 call to nack. The first send + 5 more resends + // due to the backoff setting and the above latch now counted down to zero. + // There may be a race condition, the below assertion find an extra nack, + // but the probability for that is low as we have a long enough backoff + // multiplier. + await().atMost(Duration.ofSeconds(10)) + .untilAsserted(() -> verify(containerConsumer, times(6)).negativeAcknowledge(any(Message.class))); + + container.stop(); + pulsarClient.close(); + } + } 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 3385a195..b7ea37a0 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 @@ -29,7 +29,9 @@ import java.util.concurrent.TimeUnit; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.RedeliveryBackoff; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.impl.MultiplierRedeliveryBackoff; import org.apache.pulsar.client.impl.schema.AvroSchema; import org.apache.pulsar.client.impl.schema.JSONSchema; import org.apache.pulsar.common.schema.KeyValue; @@ -226,6 +228,41 @@ public class PulsarListenerTests extends AbstractContainerBaseTests { } + @Nested + @ContextConfiguration(classes = NegativeAckRedeliveryBackoffTest.NegativeAckRedeliveryConfig.class) + class NegativeAckRedeliveryBackoffTest { + + static CountDownLatch nackRedeliveryBackoffLatch = new CountDownLatch(5); + + @Test + void pulsarListenerWithNackRedeliveryBackoff(@Autowired PulsarListenerEndpointRegistry registry) + throws Exception { + pulsarTemplate.send("withNegRedeliveryBackoff-test-topic", "hello john doe"); + assertThat(nackRedeliveryBackoffLatch.await(15, TimeUnit.SECONDS)).isTrue(); + } + + @EnablePulsar + @Configuration + static class NegativeAckRedeliveryConfig { + + @PulsarListener(id = "withNegRedeliveryBackoff", subscriptionName = "withNegRedeliveryBackoffSubscription", + topics = "withNegRedeliveryBackoff-test-topic", negativeAckRedeliveryBackoff = "redeliveryBackoff", + subscriptionType = "Shared") + void listen(String msg) { + nackRedeliveryBackoffLatch.countDown(); + throw new RuntimeException("fail " + msg); + } + + @Bean + public RedeliveryBackoff redeliveryBackoff() { + return MultiplierRedeliveryBackoff.builder().minDelayMs(1000).maxDelayMs(5 * 1000).multiplier(2) + .build(); + } + + } + + } + @Nested class NegativeConcurrency {