Nack redelivery backoff changes

For shared subscriptions, Pulsar allows consumers to provide a complex
redelivery backoff mechanism when negatively acknowledging (nack).
Enabling this Pulsar feature through the PulsarListener annotation
and its related components in the framework.

Resolves https://github.com/spring-projects-experimental/spring-pulsar/issues/78

Checkstyle cleanup

Addressing PR review

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.
 	 * <p>
-	 * 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 "";
+
 }

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<K, V>
 		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) {

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<V> extends AbstractPulsarListenerEndpo
 
 	private SmartMessageConverter messagingConverter;
 
+	private RedeliveryBackoff negativeAckRedeliveryBackoff;
+
 	public void setBean(Object bean) {
 		this.bean = bean;
 	}
@@ -172,6 +175,8 @@ public class MethodPulsarListenerEndpoint<V> 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<V> extends AbstractPulsarListenerEndpo
 		this.messagingConverter = messagingConverter;
 	}
 
+	public void setNegativeAckRedeliveryBackoff(RedeliveryBackoff negativeAckRedeliveryBackoff) {
+		this.negativeAckRedeliveryBackoff = negativeAckRedeliveryBackoff;
+	}
+
 }

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<T> implements PulsarConsumerFactory<T>
 			consumerBuilder.loadConf(properties);
 		}
 
+		if (properties.containsKey("negativeAckRedeliveryBackoff")) {
+			final RedeliveryBackoff negativeAckRedeliveryBackoff = (RedeliveryBackoff) properties
+					.get("negativeAckRedeliveryBackoff");
+			consumerBuilder.negativeAckRedeliveryBackoff(negativeAckRedeliveryBackoff);
+		}
+
 		consumerBuilder.batchReceivePolicy(batchReceivePolicy);
 		Consumer<T> consumer = consumerBuilder.subscribe();
 		this.consumers.add(consumer);

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<T> implements Pulsa
 
 	private volatile boolean running = false;
 
+	protected RedeliveryBackoff negativeAckRedeliveryBackoff;
+
 	@SuppressWarnings("unchecked")
 	protected AbstractPulsarMessageListenerContainer(PulsarConsumerFactory<? super T> pulsarConsumerFactory,
 			PulsarContainerProperties pulsarContainerProperties) {
@@ -173,4 +176,13 @@ public abstract class AbstractPulsarMessageListenerContainer<T> implements Pulsa
 		}
 	}
 
+	@Override
+	public void setNegativeAckRedeliveryBackoff(RedeliveryBackoff redeliveryBackoff) {
+		this.negativeAckRedeliveryBackoff = redeliveryBackoff;
+	}
+
+	public RedeliveryBackoff getNegativeAckRedeliveryBackoff() {
+		return this.negativeAckRedeliveryBackoff;
+	}
+
 }

spring-pulsar/src/main/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainer.java

@@ -108,6 +108,7 @@ public class ConcurrentPulsarMessageListenerContainer<T> extends AbstractPulsarM
 			this.executors.add(exec);
 			container.getContainerProperties().setConsumerTaskExecutor(exec);
 		}
+		container.setNegativeAckRedeliveryBackoff(this.negativeAckRedeliveryBackoff);
 	}
 
 	@Override
@@ -126,4 +127,8 @@ public class ConcurrentPulsarMessageListenerContainer<T> extends AbstractPulsarM
 		return false;
 	}
 
+	public List<DefaultPulsarMessageListenerContainer<T>> getContainers() {
+		return this.containers;
+	}
+
 }

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<T> extends AbstractPulsarMess
 			}
 			try {
 				final PulsarContainerProperties pulsarContainerProperties = getPulsarContainerProperties();
-				Map<String, Object> propertiesToOverride = extractPropertiesToOverride(pulsarContainerProperties);
+				Map<String, Object> 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<T> extends AbstractPulsarMess
 			}
 		}
 
-		private Map<String, Object> extractPropertiesToOverride(PulsarContainerProperties pulsarContainerProperties) {
-
+		private Map<String, Object> 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<String, Object> propOverridesAsMap = propertyOverrides.entrySet().stream().collect(Collectors
-					.toMap(e -> String.valueOf(e.getKey()), Map.Entry::getValue, (prev, next) -> next, HashMap::new));
-
-			final Map<String, Object> propertiesToOverride = new HashMap<>(propOverridesAsMap);
-			if (propertiesToOverride.containsKey("topicNames")) {
-				final String topicsFromMap = (String) propertiesToOverride.get("topicNames");
-				final String[] topicNames = topicsFromMap.split(",");
-				final Set<String> propertiesDefinedTopics = new HashSet<>(Arrays.stream(topicNames).toList());
+		private void populateAllNecessaryPropertiesIfNeedBe(Map<String, Object> currentProperties) {
+			if (currentProperties.containsKey("topicNames")) {
+				final String topicsFromMap = (String) currentProperties.get("topicNames");
+				final String[] topicNames = StringUtils.delimitedListToStringArray(topicsFromMap, ",");
+				final Set<String> 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<String> 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<T> extends AbstractPulsarMess
 			publishConsumerStartedEvent();
 			while (isRunning()) {
 				Messages<T> messages = null;
-
 				// Always receive messages in batch mode.
 				try {
 					messages = this.consumer.batchReceive();

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);
+
 }

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<String> 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<String> container = new DefaultPulsarMessageListenerContainer<>(
 				pulsarConsumerFactory, pulsarContainerProperties);
 		container.start();
-		final Consumer<?> containerConsumer = spyOnConsumer(container);
+		final Consumer<?> containerConsumer = ConsumerTestUtils.spyOnConsumer(container);
 
 		Map<String, Object> prodConfig = new HashMap<>();
 		prodConfig.put("topicName", "cons-ack-tests-012");
@@ -167,7 +164,7 @@ class ConsumerAcknowledgmentTests extends AbstractContainerBaseTests {
 		DefaultPulsarMessageListenerContainer<String> 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<String> 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<String> container = new DefaultPulsarMessageListenerContainer<>(
 				pulsarConsumerFactory, pulsarContainerProperties);
 		container.start();
-		final Consumer<?> containerConsumer = spyOnConsumer(container);
+		final Consumer<?> containerConsumer = ConsumerTestUtils.spyOnConsumer(container);
 
 		Map<String, Object> prodConfig = new HashMap<>();
 		prodConfig.put("topicName", "cons-ack-tests-015");
@@ -347,7 +344,7 @@ class ConsumerAcknowledgmentTests extends AbstractContainerBaseTests {
 		DefaultPulsarMessageListenerContainer<String> container = new DefaultPulsarMessageListenerContainer<>(
 				pulsarConsumerFactory, pulsarContainerProperties);
 		container.start();
-		final Consumer<?> containerConsumer = spyOnConsumer(container);
+		final Consumer<?> containerConsumer = ConsumerTestUtils.spyOnConsumer(container);
 
 		Map<String, Object> prodConfig = new HashMap<>();
 		prodConfig.put("topicName", "cons-ack-tests-016");
@@ -367,47 +364,4 @@ class ConsumerAcknowledgmentTests extends AbstractContainerBaseTests {
 		pulsarClient.close();
 	}
 
-	private Consumer<?> spyOnConsumer(DefaultPulsarMessageListenerContainer<String> 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> T getPropertyValue(Object root, String propertyPath, Class<T> type) {
-		Object value = getPropertyValue(root, propertyPath);
-		if (value != null) {
-			Assert.isAssignable(type, value.getClass());
-		}
-		return (T) value;
-	}
-
 }

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<String> 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> T getPropertyValue(Object root, String propertyPath, Class<T> type) {
+		Object value = getPropertyValue(root, propertyPath);
+		if (value != null) {
+			Assert.isAssignable(type, value.getClass());
+		}
+		return (T) value;
+	}
+
+}

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<String> pulsarConsumerFactory = mock(PulsarConsumerFactory.class);
+		Consumer<String> 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<String> concurrentContainer = new ConcurrentPulsarMessageListenerContainer<>(
+				pulsarConsumerFactory, pulsarContainerProperties);
+		RedeliveryBackoff redeliveryBackoff = MultiplierRedeliveryBackoff.builder().minDelayMs(1000)
+				.maxDelayMs(5 * 1000).build();
+		concurrentContainer.setNegativeAckRedeliveryBackoff(redeliveryBackoff);
+
+		concurrentContainer.start();
+
+		final DefaultPulsarMessageListenerContainer<String> childContainer = concurrentContainer.getContainers().get(0);
+		assertThat(childContainer.getNegativeAckRedeliveryBackoff()).isEqualTo(redeliveryBackoff);
+	}
+
 	@Test
 	@SuppressWarnings("unchecked")
 	void basicConcurrencyTesting() throws Exception {

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<String, Object> 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<String> 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<String> container = new DefaultPulsarMessageListenerContainer<>(
+				pulsarConsumerFactory, pulsarContainerProperties);
+		container.start();
+
+		final Consumer<?> containerConsumer = ConsumerTestUtils.spyOnConsumer(container);
+
+		Map<String, Object> prodConfig = Collections.singletonMap("topicName", "dpmlct-015");
+		final DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
+				pulsarClient, prodConfig);
+		final PulsarTemplate<String> 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();
+	}
+
 }

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 {