Reduce custom mapping in DefaultPulsarMessageListenerContainer

2022-11-03 22:36:16 -05:00
parent 5997a338bd
commit bb12a7b666
6 changed files with 300 additions and 194 deletions
--- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/ConsumerBuilderConfigurationUtil.java
+++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/ConsumerBuilderConfigurationUtil.java
@@ -0,0 +1,97 @@
+/*
+ * 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 java.util.HashMap;
+import java.util.Map;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import org.apache.pulsar.client.api.ConsumerBuilder;
+import org.apache.pulsar.client.api.DeadLetterPolicy;
+import org.apache.pulsar.client.api.RedeliveryBackoff;
+import org.apache.pulsar.client.impl.ConsumerBuilderImpl;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+
+/**
+ * Utility methods to help load configuration into a {@link ConsumerBuilder}.
+ * <p>
+ * The main purpose is to work around the underlying
+ * <a href="https://github.com/apache/pulsar/issues/11646">Pulsar issue</a> where
+ * {@code ConsumerBuilder::loadConf} sets {@code @JsonIgnore} fields to null and crashes
+ * if a {@code deadLetterPolicy} was set on the builder.
+ * <p>
+ * Should be removed once the above issue is fixed.
+ *
+ * @author Chris Bono
+ */
+public final class ConsumerBuilderConfigurationUtil {
+
+	private ConsumerBuilderConfigurationUtil() {
+	}
+
+	/**
+	 * Configures the specified properties onto the specified builder in a manner that
+	 * avoids <a href="https://github.com/apache/pulsar/issues/11646">Pulsar issue</a>.
+	 * @param builder the builder
+	 * @param properties the properties to set on the builder
+	 * @param <T> the payload type
+	 */
+	@SuppressWarnings("unchecked")
+	public static <T> void loadConf(ConsumerBuilder<T> builder, Map<String, Object> properties) {
+
+		ConsumerConfigurationData<T> builderConf = ((ConsumerBuilderImpl<T>) builder).getConf();
+		Map<String, Object> propertiesCopy = new HashMap<>(properties);
+
+		// Remove and remember problem fields from input props and builder
+		DeadLetterPolicy deadLetterPolicy = getValueToApplyToBuilderAfterLoadConf(builderConf::getDeadLetterPolicy,
+				builderConf::setDeadLetterPolicy, propertiesCopy, "deadLetterPolicy");
+		RedeliveryBackoff nackRedeliveryBackoff = getValueToApplyToBuilderAfterLoadConf(
+				builderConf::getNegativeAckRedeliveryBackoff, builderConf::setNegativeAckRedeliveryBackoff,
+				propertiesCopy, "negativeAckRedeliveryBackoff");
+		RedeliveryBackoff ackRedeliveryBackoff = getValueToApplyToBuilderAfterLoadConf(
+				builderConf::getAckTimeoutRedeliveryBackoff, builderConf::setAckTimeoutRedeliveryBackoff,
+				propertiesCopy, "ackTimeoutRedeliveryBackoff");
+
+		// DLP stripped from props - now safe to call builder.loadConf
+		builder.loadConf(propertiesCopy);
+
+		// Manually set fields marked as @JsonIgnore in ConsumerConfigurationData
+		if (deadLetterPolicy != null) {
+			builder.deadLetterPolicy(deadLetterPolicy);
+		}
+		if (nackRedeliveryBackoff != null) {
+			builder.negativeAckRedeliveryBackoff(nackRedeliveryBackoff);
+		}
+		if (ackRedeliveryBackoff != null) {
+			builder.ackTimeoutRedeliveryBackoff(ackRedeliveryBackoff);
+		}
+	}
+
+	@SuppressWarnings("unchecked")
+	private static <T> T getValueToApplyToBuilderAfterLoadConf(Supplier<T> builderGetter, Consumer<T> builderSetter,
+			Map<String, Object> properties, String propertyName) {
+		T value = (T) properties.getOrDefault(propertyName, builderGetter.get());
+
+		if ("deadLetterPolicy".equals(propertyName)) {
+			builderSetter.accept(null);
+			properties.remove(propertyName);
+		}
+		return value;
+	}
+
+}
--- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java
+++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java
@@ -27,10 +27,8 @@ import java.util.TreeMap;

 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.ConsumerBuilder;
-import org.apache.pulsar.client.api.DeadLetterPolicy;
 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.lang.Nullable;
@@ -67,7 +65,8 @@ public class DefaultPulsarConsumerFactory<T> implements PulsarConsumerFactory<T>
 	}

 	@Override
-	public Consumer<T> createConsumer(Schema<T> schema, Collection<String> topics) throws PulsarClientException {
+	public Consumer<T> createConsumer(Schema<T> schema, @Nullable Collection<String> topics)
+			throws PulsarClientException {
 		return createConsumer(schema, topics, null, Collections.emptyList());
 	}

@@ -85,30 +84,9 @@ public class DefaultPulsarConsumerFactory<T> implements PulsarConsumerFactory<T>
 			config.put("properties", new TreeMap<>(properties));
 		}

-		// Remove deadLetterPolicy from the properties here and save it to re-apply after
-		// calling `loadConf` (https://github.com/apache/pulsar/issues/11646)
-		DeadLetterPolicy deadLetterPolicy = null;
-		if (config.containsKey("deadLetterPolicy")) {
-			deadLetterPolicy = (DeadLetterPolicy) config.remove("deadLetterPolicy");
-		}
-
-		consumerBuilder.loadConf(config);
-
-		if (deadLetterPolicy != null) {
-			consumerBuilder.deadLetterPolicy(deadLetterPolicy);
-		}
-
-		if (config.containsKey("negativeAckRedeliveryBackoff")) {
-			RedeliveryBackoff negativeAckRedeliveryBackoff = (RedeliveryBackoff) config
-					.get("negativeAckRedeliveryBackoff");
-			consumerBuilder.negativeAckRedeliveryBackoff(negativeAckRedeliveryBackoff);
-		}
-
-		if (config.containsKey("ackTimeoutRedeliveryBackoff")) {
-			RedeliveryBackoff ackTimeoutRedeliveryBackoff = (RedeliveryBackoff) config
-					.get("ackTimeoutRedeliveryBackoff");
-			consumerBuilder.ackTimeoutRedeliveryBackoff(ackTimeoutRedeliveryBackoff);
-		}
+		// Replace w/ consumerBuilder.loadConf after
+		// https://github.com/apache/pulsar/issues/11646
+		ConsumerBuilderConfigurationUtil.loadConf(consumerBuilder, config);

 		if (!CollectionUtils.isEmpty(customizers)) {
 			customizers.forEach(customizer -> customizer.customize(consumerBuilder));
--- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java
+++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java
@@ -33,6 +33,7 @@ import org.springframework.lang.Nullable;
 * @param <T> payload type for the consumer.
 * @author Soby Chacko
 * @author Christophe Bornet
+ * @author Chris Bono
 */
 public interface PulsarConsumerFactory<T> {

@@ -47,11 +48,12 @@ public interface PulsarConsumerFactory<T> {
 	/**
 	 * Create a consumer.
 	 * @param schema the schema of the messages to be sent
-	 * @param topics the topics the consumer will subscribe to
+	 * @param topics the topics the consumer will subscribe to overriding the default ones
+	 * or {@code null} to use the default topics
 	 * @return the consumer
 	 * @throws PulsarClientException if any error occurs
 	 */
-	Consumer<T> createConsumer(Schema<T> schema, Collection<String> topics) throws PulsarClientException;
+	Consumer<T> createConsumer(Schema<T> schema, @Nullable Collection<String> topics) throws PulsarClientException;

 	/**
 	 * Create a consumer.
@@ -60,10 +62,11 @@ public interface PulsarConsumerFactory<T> {
 	 * or {@code null} to use the default topics. Beware that using
 	 * {@link ConsumerBuilder#topic} or {@link ConsumerBuilder#topics} will add to the
 	 * default topics, not override them.
-	 * @param properties the properties to set to the consumer overriding the default ones
-	 * or {@code null} to use the default properties. Beware that using
-	 * {@link ConsumerBuilder#property} or {@link ConsumerBuilder#properties} will add to
-	 * the default properties, not override them.
+	 * @param properties the metadata properties to attach to the consumer, replacing the
+	 * default metadata properties, or {@code null} to use the default metadata
+	 * properties. Beware that using {@link ConsumerBuilder#property} or
+	 * {@link ConsumerBuilder#properties} will add to the default metadata properties, not
+	 * replace them.
 	 * @param customizers the optional list of customizers to apply to the consumer
 	 * builder
 	 * @return the consumer
--- 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,24 +37,14 @@ import java.util.stream.StreamSupport;
 import org.apache.commons.logging.LogFactory;
 import org.apache.pulsar.client.api.BatchReceivePolicy;
 import org.apache.pulsar.client.api.Consumer;
-import org.apache.pulsar.client.api.ConsumerBuilder;
-import org.apache.pulsar.client.api.ConsumerCryptoFailureAction;
-import org.apache.pulsar.client.api.ConsumerEventListener;
-import org.apache.pulsar.client.api.CryptoKeyReader;
 import org.apache.pulsar.client.api.DeadLetterPolicy;
-import org.apache.pulsar.client.api.KeySharedPolicy;
 import org.apache.pulsar.client.api.Message;
-import org.apache.pulsar.client.api.MessageCrypto;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.MessageListener;
-import org.apache.pulsar.client.api.MessagePayloadProcessor;
 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.RegexSubscriptionMode;
 import org.apache.pulsar.client.api.Schema;
-import org.apache.pulsar.client.api.SubscriptionInitialPosition;
-import org.apache.pulsar.client.api.SubscriptionMode;
 import org.apache.pulsar.client.api.SubscriptionType;

 import org.springframework.context.ApplicationEventPublisher;
@@ -62,6 +52,7 @@ import org.springframework.core.log.LogAccessor;
 import org.springframework.core.task.AsyncTaskExecutor;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.lang.Nullable;
+import org.springframework.pulsar.core.ConsumerBuilderConfigurationUtil;
 import org.springframework.pulsar.core.ConsumerBuilderCustomizer;
 import org.springframework.pulsar.core.PulsarConsumerFactory;
 import org.springframework.pulsar.event.ConsumerFailedToStartEvent;
@@ -263,7 +254,9 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
 				Map<String, String> properties = (Map<String, String>) propertiesToConsumer.remove("properties");

 				ConsumerBuilderCustomizer<T> customizer = builder -> {
-					loadConf(builder, propertiesToConsumer);
+					// Replace w/ consumerBuilder.loadConf after
+					// https://github.com/apache/pulsar/issues/11646
+					ConsumerBuilderConfigurationUtil.loadConf(builder, propertiesToConsumer);
 					builder.batchReceivePolicy(batchReceivePolicy);
 				};
 				this.consumer = getPulsarConsumerFactory().createConsumer((Schema) containerProperties.getSchema(),
@@ -275,155 +268,6 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
 			}
 		}

-		/*
-		 * Method to work around the issue that ConsumerBuilder::loadConf crashes if a
-		 * deadLetterPolicy was set on the builder. To be removed once
-		 * https://github.com/apache/pulsar/issues/11646 is fixed.
-		 */
-		@SuppressWarnings("unchecked")
-		private static <T> void loadConf(ConsumerBuilder<T> builder, Map<String, Object> properties) {
-			if (properties.containsKey("topicsPattern")) {
-				builder.topicsPattern(properties.get("topicsPattern").toString());
-			}
-			if (properties.containsKey("subscriptionName")) {
-				builder.subscriptionName(properties.get("subscriptionName").toString());
-			}
-			if (properties.containsKey("subscriptionType")) {
-				builder.subscriptionType(SubscriptionType.valueOf(properties.get("subscriptionType").toString()));
-			}
-			if (properties.containsKey("subscriptionMode")) {
-				builder.subscriptionMode(SubscriptionMode.valueOf(properties.get("subscriptionMode").toString()));
-			}
-			if (properties.containsKey("subscriptionProperties")) {
-				builder.subscriptionProperties((Map<String, String>) properties.get("subscriptionProperties"));
-			}
-			if (properties.containsKey("messageListener")) {
-				builder.messageListener((MessageListener<T>) properties.get("messageListener"));
-			}
-			if (properties.containsKey("consumerEventListener")) {
-				builder.consumerEventListener((ConsumerEventListener) properties.get("consumerEventListener"));
-			}
-			if (properties.containsKey("negativeAckRedeliveryBackoff")) {
-				builder.negativeAckRedeliveryBackoff(
-						(RedeliveryBackoff) properties.get("negativeAckRedeliveryBackoff"));
-			}
-			if (properties.containsKey("ackTimeoutRedeliveryBackoff")) {
-				builder.ackTimeoutRedeliveryBackoff((RedeliveryBackoff) properties.get("ackTimeoutRedeliveryBackoff"));
-			}
-			if (properties.containsKey("receiverQueueSize")) {
-				builder.receiverQueueSize(Integer.parseInt(properties.get("receiverQueueSize").toString()));
-			}
-			if (properties.containsKey("acknowledgementsGroupTimeMicros")) {
-				builder.acknowledgmentGroupTime(
-						Long.parseLong(properties.get("acknowledgementsGroupTimeMicros").toString()),
-						TimeUnit.MICROSECONDS);
-			}
-			if (properties.containsKey("negativeAckRedeliveryDelayMicros")) {
-				builder.negativeAckRedeliveryDelay(
-						Long.parseLong(properties.get("negativeAckRedeliveryDelayMicros").toString()),
-						TimeUnit.MICROSECONDS);
-			}
-			if (properties.containsKey("maxTotalReceiverQueueSizeAcrossPartitions")) {
-				builder.maxTotalReceiverQueueSizeAcrossPartitions(
-						Integer.parseInt(properties.get("maxTotalReceiverQueueSizeAcrossPartitions").toString()));
-			}
-			if (properties.containsKey("consumerName")) {
-				builder.consumerName(properties.get("consumerName").toString());
-			}
-			if (properties.containsKey("ackTimeoutMillis")) {
-				builder.ackTimeout(Long.parseLong(properties.get("ackTimeoutMillis").toString()),
-						TimeUnit.MILLISECONDS);
-			}
-			if (properties.containsKey("tickDurationMillis")) {
-				builder.ackTimeoutTickTime(Long.parseLong(properties.get("tickDurationMillis").toString()),
-						TimeUnit.MILLISECONDS);
-			}
-			if (properties.containsKey("priorityLevel")) {
-				builder.priorityLevel(Integer.parseInt(properties.get("priorityLevel").toString()));
-			}
-			if (properties.containsKey("maxPendingChunkedMessage")) {
-				builder.maxPendingChunkedMessage(
-						Integer.parseInt(properties.get("maxPendingChunkedMessage").toString()));
-			}
-			if (properties.containsKey("autoAckOldestChunkedMessageOnQueueFull")) {
-				builder.autoAckOldestChunkedMessageOnQueueFull(
-						Boolean.parseBoolean(properties.get("autoAckOldestChunkedMessageOnQueueFull").toString()));
-			}
-			if (properties.containsKey("expireTimeOfIncompleteChunkedMessageMillis")) {
-				builder.expireTimeOfIncompleteChunkedMessage(
-						Long.parseLong(properties.get("expireTimeOfIncompleteChunkedMessageMillis").toString()),
-						TimeUnit.MILLISECONDS);
-			}
-			if (properties.containsKey("cryptoKeyReader")) {
-				builder.cryptoKeyReader((CryptoKeyReader) properties.get("cryptoKeyReader"));
-			}
-			if (properties.containsKey("messageCrypto")) {
-				builder.messageCrypto((MessageCrypto) properties.get("messageCrypto"));
-			}
-			if (properties.containsKey("cryptoFailureAction")) {
-				builder.cryptoFailureAction(
-						ConsumerCryptoFailureAction.valueOf(properties.get("cryptoFailureAction").toString()));
-			}
-			if (properties.containsKey("readCompacted")) {
-				builder.readCompacted(Boolean.parseBoolean(properties.get("readCompacted").toString()));
-			}
-			if (properties.containsKey("subscriptionInitialPosition")) {
-				builder.subscriptionInitialPosition(
-						SubscriptionInitialPosition.valueOf(properties.get("subscriptionInitialPosition").toString()));
-			}
-			if (properties.containsKey("patternAutoDiscoveryPeriod")) {
-				builder.patternAutoDiscoveryPeriod(
-						Integer.parseInt(properties.get("patternAutoDiscoveryPeriod").toString()), TimeUnit.SECONDS);
-			}
-			if (properties.containsKey("regexSubscriptionMode")) {
-				builder.subscriptionTopicsMode(
-						RegexSubscriptionMode.valueOf(properties.get("regexSubscriptionMode").toString()));
-			}
-			if (properties.containsKey("deadLetterPolicy")) {
-				builder.deadLetterPolicy((DeadLetterPolicy) properties.get("deadLetterPolicy"));
-			}
-			if (properties.containsKey("retryEnable")) {
-				builder.enableRetry(Boolean.parseBoolean(properties.get("retryEnable").toString()));
-			}
-			if (properties.containsKey("batchReceivePolicy")) {
-				builder.batchReceivePolicy((BatchReceivePolicy) properties.get("batchReceivePolicy"));
-			}
-			if (properties.containsKey("autoUpdatePartitions")) {
-				builder.autoUpdatePartitions(Boolean.parseBoolean(properties.get("autoUpdatePartitions").toString()));
-			}
-			if (properties.containsKey("autoUpdatePartitionsIntervalSeconds")) {
-				builder.autoUpdatePartitionsInterval(
-						Integer.parseInt(properties.get("autoUpdatePartitionsIntervalSeconds").toString()),
-						TimeUnit.SECONDS);
-			}
-			if (properties.containsKey("replicateSubscriptionState")) {
-				builder.replicateSubscriptionState(
-						Boolean.parseBoolean(properties.get("replicateSubscriptionState").toString()));
-			}
-			if (properties.containsKey("resetIncludeHead")) {
-				builder.startMessageIdInclusive();
-			}
-			if (properties.containsKey("keySharedPolicy")) {
-				builder.keySharedPolicy((KeySharedPolicy) properties.get("keySharedPolicy"));
-			}
-			if (properties.containsKey("batchIndexAckEnabled")) {
-				builder.enableBatchIndexAcknowledgment(
-						Boolean.parseBoolean(properties.get("batchIndexAckEnabled").toString()));
-			}
-			if (properties.containsKey("ackReceiptEnabled")) {
-				builder.isAckReceiptEnabled(Boolean.parseBoolean(properties.get("ackReceiptEnabled").toString()));
-			}
-			if (properties.containsKey("poolMessages")) {
-				builder.poolMessages(Boolean.parseBoolean(properties.get("poolMessages").toString()));
-			}
-			if (properties.containsKey("payloadProcessor")) {
-				builder.messagePayloadProcessor((MessagePayloadProcessor) properties.get("payloadProcessor"));
-			}
-			if (properties.containsKey("startPaused")) {
-				builder.startPaused(Boolean.parseBoolean(properties.get("startPaused").toString()));
-			}
-		}
-
 		private Map<String, Object> extractDirectConsumerProperties() {
 			Properties propertyOverrides = this.containerProperties.getPulsarConsumerProperties();
 			return propertyOverrides.entrySet().stream().collect(Collectors.toMap(e -> String.valueOf(e.getKey()),
--- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerBuilderConfigurationUtilTests.java
+++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerBuilderConfigurationUtilTests.java
@@ -0,0 +1,122 @@
+/*
+ * 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.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import org.apache.pulsar.client.api.ConsumerBuilder;
+import org.apache.pulsar.client.api.DeadLetterPolicy;
+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.apache.pulsar.client.impl.MultiplierRedeliveryBackoff;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.assertj.core.api.InstanceOfAssertFactories;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import org.springframework.lang.Nullable;
+import org.springframework.test.util.ReflectionTestUtils;
+
+/**
+ * Unit tests for {@link ConsumerBuilderConfigurationUtil}.
+ *
+ * @author Chris Bono
+ */
+public class ConsumerBuilderConfigurationUtilTests {
+
+	private ConsumerBuilder<String> builder;
+
+	@BeforeEach
+	void prepareBuilder() throws PulsarClientException {
+		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl("pulsar://localhost:6650").build();
+		builder = pulsarClient.newConsumer(Schema.STRING);
+	}
+
+	@ParameterizedTest(name = "{0}")
+	@MethodSource("loadConfTestProvider")
+	void loadConfTest(String testName, String propName, @Nullable Object objOnBuilder, @Nullable Object objOnProps,
+			@Nullable Object expectedObj) {
+
+		if (objOnBuilder != null) {
+			ReflectionTestUtils.invokeSetterMethod(builder, propName, objOnBuilder);
+		}
+
+		Map<String, Object> props = new HashMap<>();
+		if (objOnProps != null) {
+			props.put(propName, objOnProps);
+		}
+		Map<String, Object> propsBeforeUtil = new HashMap<>(props);
+
+		ConsumerBuilderConfigurationUtil.loadConf(builder, props);
+
+		assertThat(this.builder).extracting("conf")
+				.asInstanceOf(InstanceOfAssertFactories.type(ConsumerConfigurationData.class)).extracting(propName)
+				.isEqualTo(expectedObj);
+
+		assertThat(props).isEqualTo(propsBeforeUtil);
+	}
+
+	private static Stream<Arguments> loadConfTestProvider() {
+
+		DeadLetterPolicy deadLetterPolicyOnBuilder = DeadLetterPolicy.builder().deadLetterTopic("dlt-topic")
+				.maxRedeliverCount(1).build();
+		DeadLetterPolicy deadLetterPolicyInProps = DeadLetterPolicy.builder().deadLetterTopic("dlt-topic")
+				.maxRedeliverCount(2).build();
+
+		RedeliveryBackoff nackRedeliveryBackoffOnBuilder = MultiplierRedeliveryBackoff.builder().minDelayMs(1000)
+				.maxDelayMs(5000).build();
+		RedeliveryBackoff nackRedeliveryBackoffInProps = MultiplierRedeliveryBackoff.builder().minDelayMs(2000)
+				.maxDelayMs(4000).build();
+
+		RedeliveryBackoff ackRedeliveryBackoffOnBuilder = MultiplierRedeliveryBackoff.builder().minDelayMs(1000)
+				.maxDelayMs(5000).build();
+		RedeliveryBackoff ackRedeliveryBackoffInProps = MultiplierRedeliveryBackoff.builder().minDelayMs(2000)
+				.maxDelayMs(4000).build();
+
+		return Stream.of(arguments("loadConfNoDeadLetterPolicy", "deadLetterPolicy", null, null, null),
+				arguments("loadConfDeadLetterPolicyOnBuilder", "deadLetterPolicy", deadLetterPolicyOnBuilder, null,
+						deadLetterPolicyOnBuilder),
+				arguments("loadConfDeadLetterPolicyInProps", "deadLetterPolicy", null, deadLetterPolicyInProps,
+						deadLetterPolicyInProps),
+				arguments("loadConfDeadLetterPolicyOnBuilderAndInProps", "deadLetterPolicy", deadLetterPolicyOnBuilder,
+						deadLetterPolicyInProps, deadLetterPolicyInProps),
+				arguments("loadConfNoNegativeAckRedeliveryBackoff", "negativeAckRedeliveryBackoff", null, null, null),
+				arguments("loadConfNegativeAckRedeliveryBackoffOnBuilder", "negativeAckRedeliveryBackoff",
+						nackRedeliveryBackoffOnBuilder, null, nackRedeliveryBackoffOnBuilder),
+				arguments("loadConfNegativeAckRedeliveryBackoffInProps", "negativeAckRedeliveryBackoff", null,
+						nackRedeliveryBackoffInProps, nackRedeliveryBackoffInProps),
+				arguments("loadConfNegativeAckRedeliveryBackoffOnBuilderAndInProps", "negativeAckRedeliveryBackoff",
+						nackRedeliveryBackoffOnBuilder, nackRedeliveryBackoffInProps, nackRedeliveryBackoffInProps),
+				arguments("loadConfNoAckRedeliveryBackoff", "ackTimeoutRedeliveryBackoff", null, null, null),
+				arguments("loadConfAckRedeliveryBackoffOnBuilder", "ackTimeoutRedeliveryBackoff",
+						ackRedeliveryBackoffOnBuilder, null, ackRedeliveryBackoffOnBuilder),
+				arguments("loadConfAckRedeliveryBackoffInProps", "ackTimeoutRedeliveryBackoff", null,
+						ackRedeliveryBackoffInProps, ackRedeliveryBackoffInProps),
+				arguments("loadConfAckRedeliveryBackoffOnBuilderAndInProps", "ackTimeoutRedeliveryBackoff",
+						ackRedeliveryBackoffOnBuilder, ackRedeliveryBackoffInProps, ackRedeliveryBackoffInProps));
+	}
+
+}
--- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java
+++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java
@@ -52,6 +52,7 @@ import org.springframework.pulsar.core.PulsarTestContainerSupport;
 /**
 * @author Soby Chacko
 * @author Alexander Preuß
+ * @author Chris Bono
 */
 class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerSupport {

@@ -209,7 +210,7 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
 	}

 	@Test
-	void deadLetterPolicy() throws Exception {
+	void deadLetterPolicyDefault() throws Exception {
 		Map<String, Object> config = new HashMap<>();
 		config.put("topicNames", Collections.singleton("dpmlct-016"));
 		config.put("subscriptionName", "dpmlct-sb-016");
@@ -266,4 +267,65 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
 		pulsarClient.close();
 	}

+	@Test
+	void deadLetterPolicyCustom() throws Exception {
+		Map<String, Object> config = new HashMap<>();
+		config.put("topicNames", Collections.singleton("dpmlct-016"));
+		config.put("subscriptionName", "dpmlct-sb-016");
+		config.put("ackTimeoutMillis", 1);
+		DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder().maxRedeliverCount(5).deadLetterTopic("dlq-topic")
+				.build();
+		config.put("deadLetterPolicy", deadLetterPolicy);
+
+		final PulsarClient pulsarClient = PulsarClient.builder()
+				.serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
+		final DefaultPulsarConsumerFactory<Integer> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(
+				pulsarClient, config);
+
+		CountDownLatch dlqLatch = new CountDownLatch(1);
+		CountDownLatch latch = new CountDownLatch(6);
+
+		PulsarContainerProperties dlqContainerProperties = new PulsarContainerProperties();
+		dlqContainerProperties
+				.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> dlqLatch.countDown());
+		dlqContainerProperties.setSchema(Schema.INT32);
+		dlqContainerProperties.setSubscriptionType(SubscriptionType.Shared);
+		dlqContainerProperties.setTopics(new String[] { "dlq-topic" });
+		DefaultPulsarMessageListenerContainer<Integer> dlqContainer = new DefaultPulsarMessageListenerContainer<>(
+				pulsarConsumerFactory, dlqContainerProperties);
+		dlqContainer.start();
+
+		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
+		pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener<Integer>) (consumer, msg) -> {
+			latch.countDown();
+			if (msg.getValue() == 5) {
+				throw new RuntimeException("fail");
+			}
+		});
+		pulsarContainerProperties.setSchema(Schema.INT32);
+		pulsarContainerProperties.setSubscriptionType(SubscriptionType.Shared);
+		pulsarContainerProperties.getPulsarConsumerProperties().put("deadLetterPolicy",
+				DeadLetterPolicy.builder().maxRedeliverCount(1).deadLetterTopic("dlq-topic").build());
+
+		DefaultPulsarMessageListenerContainer<Integer> container = new DefaultPulsarMessageListenerContainer<>(
+				pulsarConsumerFactory, pulsarContainerProperties);
+		container.start();
+
+		Map<String, Object> prodConfig = Collections.singletonMap("topicName", "dpmlct-016");
+		final DefaultPulsarProducerFactory<Integer> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
+				pulsarClient, prodConfig);
+		final PulsarTemplate<Integer> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
+		for (int i = 1; i < 6; i++) {
+			pulsarTemplate.send(i);
+		}
+
+		// DLQ consumer should receive 1 msg
+		assertThat(dlqLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		// Normal consumer should receive 5 msg + 1 re-delivery
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		container.stop();
+		dlqContainer.stop();
+		pulsarClient.close();
+	}
+
 }