Upgrade to pulsar client 2.10.2

spring-pulsar-dependencies/build.gradle

@@ -12,7 +12,7 @@ ext {
 	micrometerTracingVersion = '1.0.0'
 	protobufJavaVersion = '3.21.5'
 	testcontainersVersion = '1.17.3'
-	pulsarVersion = '2.10.1'
+	pulsarVersion = '2.10.2'
 	pulsarClientReactiveVersion = '0.1.0-SNAPSHOT'
 	springBootVersion = '3.0.0-RC1'
 }

spring-pulsar/src/main/java/org/springframework/pulsar/core/ConsumerBuilderConfigurationUtil.java

@@ -22,8 +22,6 @@ 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;
 
@@ -31,9 +29,8 @@ 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.
+ * <a href="https://github.com/apache/pulsar/pull/18344">Pulsar issue</a> where
+ * {@link ConsumerBuilder#loadConf} sets {@code @JsonIgnore} fields to null.
  * <p>
  * Should be removed once the above issue is fixed.
  *
@@ -46,52 +43,53 @@ public final class 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>.
+	 * loads non-serializable properties. See
+	 * <a href="https://github.com/apache/pulsar/pull/18344">Pulsar PR</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);
+		propertiesCopy.remove("messageListener");
+		propertiesCopy.remove("consumerEventListener");
+		propertiesCopy.remove("negativeAckRedeliveryBackoff");
+		propertiesCopy.remove("ackTimeoutRedeliveryBackoff");
+		propertiesCopy.remove("cryptoKeyReader");
+		propertiesCopy.remove("messageCrypto");
+		propertiesCopy.remove("batchReceivePolicy");
+		propertiesCopy.remove("payloadProcessor");
 
-		// 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);
-		}
+		applyValueToBuilderAfterLoadConf(builderConf::getMessageListener, builder::messageListener, properties,
+				"messageListener");
+		applyValueToBuilderAfterLoadConf(builderConf::getConsumerEventListener, builder::consumerEventListener,
+				properties, "consumerEventListener");
+		applyValueToBuilderAfterLoadConf(builderConf::getNegativeAckRedeliveryBackoff,
+				builder::negativeAckRedeliveryBackoff, properties, "negativeAckRedeliveryBackoff");
+		applyValueToBuilderAfterLoadConf(builderConf::getAckTimeoutRedeliveryBackoff,
+				builder::ackTimeoutRedeliveryBackoff, properties, "ackTimeoutRedeliveryBackoff");
+		applyValueToBuilderAfterLoadConf(builderConf::getCryptoKeyReader, builder::cryptoKeyReader, properties,
+				"cryptoKeyReader");
+		applyValueToBuilderAfterLoadConf(builderConf::getMessageCrypto, builder::messageCrypto, properties,
+				"messageCrypto");
+		applyValueToBuilderAfterLoadConf(builderConf::getBatchReceivePolicy, builder::batchReceivePolicy, properties,
+				"batchReceivePolicy");
+		applyValueToBuilderAfterLoadConf(builderConf::getPayloadProcessor, builder::messagePayloadProcessor, properties,
+				"payloadProcessor");
 	}
 
 	@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);
+	private static <T> void applyValueToBuilderAfterLoadConf(Supplier<T> confGetter, Consumer<T> builderSetter,
+			Map<String, Object> properties, String key) {
+		T value = (T) properties.getOrDefault(key, confGetter.get());
+		if (value != null) {
+			builderSetter.accept(value);
 		}
-		return value;
 	}
 
 }

spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java

@@ -84,8 +84,6 @@ public class DefaultPulsarConsumerFactory<T> implements PulsarConsumerFactory<T>
 			config.put("properties", new TreeMap<>(properties));
 		}
 
-		// Replace w/ consumerBuilder.loadConf after
-		// https://github.com/apache/pulsar/issues/11646
 		ConsumerBuilderConfigurationUtil.loadConf(consumerBuilder, config);
 
 		if (!CollectionUtils.isEmpty(customizers)) {

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

@@ -254,8 +254,6 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
 				Map<String, String> properties = (Map<String, String>) propertiesToConsumer.remove("properties");
 
 				ConsumerBuilderCustomizer<T> customizer = builder -> {
-					// Replace w/ consumerBuilder.loadConf after
-					// https://github.com/apache/pulsar/issues/11646
 					ConsumerBuilderConfigurationUtil.loadConf(builder, propertiesToConsumer);
 					builder.batchReceivePolicy(batchReceivePolicy);
 				};

spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java

@@ -24,6 +24,7 @@ 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;
 
@@ -69,8 +70,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 		config.put("subscriptionName", "cons-ack-tests-sb-011");
 		final PulsarClient pulsarClient = PulsarClient.builder()
 				.serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
-		final DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(
-				pulsarClient, config);
+		final DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(
+				new DefaultPulsarConsumerFactory<>(pulsarClient, config));
 
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
@@ -79,8 +80,7 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 		pulsarContainerProperties.setAckMode(AckMode.RECORD);
 		DefaultPulsarMessageListenerContainer<String> container = new DefaultPulsarMessageListenerContainer<>(
 				pulsarConsumerFactory, pulsarContainerProperties);
-		container.start();
-		final Consumer<?> containerConsumer = ConsumerTestUtils.spyOnConsumer(container);
+		final Consumer<String> containerConsumer = ConsumerTestUtils.startContainerAndSpyOnConsumer(container);
 
 		CountDownLatch latch = new CountDownLatch(10);
 
@@ -111,8 +111,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 		config.put("subscriptionName", "cons-ack-tests-sb-012");
 		final PulsarClient pulsarClient = PulsarClient.builder()
 				.serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
-		final DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(
-				pulsarClient, config);
+		final DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(
+				new DefaultPulsarConsumerFactory<>(pulsarClient, config));
 
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		CountDownLatch latch = new CountDownLatch(10);
@@ -121,8 +121,7 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 		pulsarContainerProperties.setSchema(Schema.STRING);
 		DefaultPulsarMessageListenerContainer<String> container = new DefaultPulsarMessageListenerContainer<>(
 				pulsarConsumerFactory, pulsarContainerProperties);
-		container.start();
-		final Consumer<?> containerConsumer = ConsumerTestUtils.spyOnConsumer(container);
+		final Consumer<String> containerConsumer = ConsumerTestUtils.startContainerAndSpyOnConsumer(container);
 
 		Map<String, Object> prodConfig = new HashMap<>();
 		prodConfig.put("topicName", "cons-ack-tests-012");
@@ -148,8 +147,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 		config.put("subscriptionName", "cons-ack-tests-sb-013");
 		final PulsarClient pulsarClient = PulsarClient.builder()
 				.serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
-		final DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(
-				pulsarClient, config);
+		final DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(
+				new DefaultPulsarConsumerFactory<>(pulsarClient, config));
 
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		CountDownLatch latch = new CountDownLatch(10);
@@ -163,8 +162,7 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 		pulsarContainerProperties.setSchema(Schema.STRING);
 		DefaultPulsarMessageListenerContainer<String> container = new DefaultPulsarMessageListenerContainer<>(
 				pulsarConsumerFactory, pulsarContainerProperties);
-		container.start();
-		final Consumer<?> containerConsumer = ConsumerTestUtils.spyOnConsumer(container);
+		final Consumer<String> containerConsumer = ConsumerTestUtils.startContainerAndSpyOnConsumer(container);
 
 		AtomicInteger ackCallCount = new AtomicInteger(0);
 		doAnswer(invocation -> {
@@ -217,8 +215,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 		config.put("subscriptionName", "cons-ack-tests-sb-014");
 		final PulsarClient pulsarClient = PulsarClient.builder()
 				.serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
-		final DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(
-				pulsarClient, config);
+		final DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(
+				new DefaultPulsarConsumerFactory<>(pulsarClient, config));
 
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		final List<Acknowledgement> acksObjects = new ArrayList<>();
@@ -233,8 +231,7 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 		pulsarContainerProperties.setAckMode(AckMode.MANUAL);
 		DefaultPulsarMessageListenerContainer<String> container = new DefaultPulsarMessageListenerContainer<>(
 				pulsarConsumerFactory, pulsarContainerProperties);
-		container.start();
-		final Consumer<?> containerConsumer = ConsumerTestUtils.spyOnConsumer(container);
+		final Consumer<String> containerConsumer = ConsumerTestUtils.startContainerAndSpyOnConsumer(container);
 
 		CountDownLatch latch = new CountDownLatch(10);
 
@@ -272,8 +269,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 		config.put("subscriptionName", "cons-ack-tests-sb-015");
 		final PulsarClient pulsarClient = PulsarClient.builder()
 				.serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
-		final DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(
-				pulsarClient, config);
+		final DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(
+				new DefaultPulsarConsumerFactory<>(pulsarClient, config));
 
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		pulsarContainerProperties.setMaxNumMessages(10);
@@ -291,8 +288,7 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 		pulsarContainerProperties.setSchema(Schema.STRING);
 		DefaultPulsarMessageListenerContainer<String> container = new DefaultPulsarMessageListenerContainer<>(
 				pulsarConsumerFactory, pulsarContainerProperties);
-		container.start();
-		final Consumer<?> containerConsumer = ConsumerTestUtils.spyOnConsumer(container);
+		final Consumer<String> containerConsumer = ConsumerTestUtils.startContainerAndSpyOnConsumer(container);
 
 		Map<String, Object> prodConfig = new HashMap<>();
 		prodConfig.put("topicName", "cons-ack-tests-015");
@@ -321,8 +317,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 		config.put("subscriptionName", "cons-ack-tests-sb-016");
 		final PulsarClient pulsarClient = PulsarClient.builder()
 				.serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
-		final DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(
-				pulsarClient, config);
+		final DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(
+				new DefaultPulsarConsumerFactory<>(pulsarClient, config));
 
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		pulsarContainerProperties.setMaxNumMessages(10);
@@ -340,8 +336,7 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 		pulsarContainerProperties.setSchema(Schema.STRING);
 		DefaultPulsarMessageListenerContainer<String> container = new DefaultPulsarMessageListenerContainer<>(
 				pulsarConsumerFactory, pulsarContainerProperties);
-		container.start();
-		final Consumer<?> containerConsumer = ConsumerTestUtils.spyOnConsumer(container);
+		final Consumer<String> containerConsumer = ConsumerTestUtils.startContainerAndSpyOnConsumer(container);
 
 		Map<String, Object> prodConfig = new HashMap<>();
 		prodConfig.put("topicName", "cons-ack-tests-016");

spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerBuilderConfigurationUtilTests.java

@@ -18,18 +18,24 @@ package org.springframework.pulsar.core;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.params.provider.Arguments.arguments;
+import static org.mockito.Mockito.mock;
 
 import java.util.HashMap;
 import java.util.Map;
 import java.util.stream.Stream;
 
+import org.apache.pulsar.client.api.BatchReceivePolicy;
 import org.apache.pulsar.client.api.ConsumerBuilder;
+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.MessageCrypto;
+import org.apache.pulsar.client.api.MessageListener;
+import org.apache.pulsar.client.api.MessagePayloadProcessor;
 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;
@@ -38,7 +44,6 @@ 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}.
@@ -57,11 +62,11 @@ public class ConsumerBuilderConfigurationUtilTests {
 
 	@ParameterizedTest(name = "{0}")
 	@MethodSource("loadConfTestProvider")
-	void loadConfTest(String testName, String propName, @Nullable Object objOnBuilder, @Nullable Object objOnProps,
-			@Nullable Object expectedObj) {
+	void loadConfTest(String testName, String propName, @Nullable ConsumerBuilderCustomizer<String> objOnBuilder,
+			@Nullable Object objOnProps, @Nullable Object expectedObj) {
 
 		if (objOnBuilder != null) {
-			ReflectionTestUtils.invokeSetterMethod(builder, propName, objOnBuilder);
+			objOnBuilder.customize(builder);
 		}
 
 		Map<String, Object> props = new HashMap<>();
@@ -79,44 +84,116 @@ public class ConsumerBuilderConfigurationUtilTests {
 		assertThat(props).isEqualTo(propsBeforeUtil);
 	}
 
+	@SuppressWarnings({ "unchecked", "rawtypes" })
 	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();
+		ConsumerBuilderCustomizer<String> deadLetterPolicyCustomizer = c -> c
+				.deadLetterPolicy(deadLetterPolicyOnBuilder);
 
-		RedeliveryBackoff nackRedeliveryBackoffOnBuilder = MultiplierRedeliveryBackoff.builder().minDelayMs(1000)
-				.maxDelayMs(5000).build();
-		RedeliveryBackoff nackRedeliveryBackoffInProps = MultiplierRedeliveryBackoff.builder().minDelayMs(2000)
-				.maxDelayMs(4000).build();
+		MessageListener<String> messageListenerOnBuilder = mock(MessageListener.class);
+		MessageListener<String> messageListenerInProps = mock(MessageListener.class);
+		ConsumerBuilderCustomizer<String> messageListenerCustomizer = c -> c.messageListener(messageListenerOnBuilder);
 
-		RedeliveryBackoff ackRedeliveryBackoffOnBuilder = MultiplierRedeliveryBackoff.builder().minDelayMs(1000)
-				.maxDelayMs(5000).build();
-		RedeliveryBackoff ackRedeliveryBackoffInProps = MultiplierRedeliveryBackoff.builder().minDelayMs(2000)
-				.maxDelayMs(4000).build();
+		ConsumerEventListener consumerEventListenerOnBuilder = mock(ConsumerEventListener.class);
+		ConsumerEventListener consumerEventListenerInProps = mock(ConsumerEventListener.class);
+		ConsumerBuilderCustomizer<String> consumerEventListenerCustomizer = c -> c
+				.consumerEventListener(consumerEventListenerOnBuilder);
+
+		RedeliveryBackoff nackRedeliveryBackoffOnBuilder = mock(RedeliveryBackoff.class);
+		RedeliveryBackoff nackRedeliveryBackoffInProps = mock(RedeliveryBackoff.class);
+		ConsumerBuilderCustomizer<String> nackRedeliveryBackoffCustomizer = c -> c
+				.negativeAckRedeliveryBackoff(nackRedeliveryBackoffOnBuilder);
+
+		RedeliveryBackoff ackRedeliveryBackoffOnBuilder = mock(RedeliveryBackoff.class);
+		RedeliveryBackoff ackRedeliveryBackoffInProps = mock(RedeliveryBackoff.class);
+		ConsumerBuilderCustomizer<String> ackRedeliveryBackoffCustomizer = c -> c
+				.ackTimeoutRedeliveryBackoff(ackRedeliveryBackoffOnBuilder);
+
+		CryptoKeyReader cryptoKeyReaderOnBuilder = mock(CryptoKeyReader.class);
+		CryptoKeyReader cryptoKeyReaderInProps = mock(CryptoKeyReader.class);
+		ConsumerBuilderCustomizer<String> cryptoKeyReaderCustomizer = c -> c.cryptoKeyReader(cryptoKeyReaderOnBuilder);
+
+		MessageCrypto messageCryptoOnBuilder = mock(MessageCrypto.class);
+		MessageCrypto messageCryptoInProps = mock(MessageCrypto.class);
+		ConsumerBuilderCustomizer<String> messageCryptoCustomizer = c -> c.messageCrypto(messageCryptoOnBuilder);
+
+		BatchReceivePolicy batchReceivePolicyOnBuilder = mock(BatchReceivePolicy.class);
+		BatchReceivePolicy batchReceivePolicyInProps = mock(BatchReceivePolicy.class);
+		ConsumerBuilderCustomizer<String> batchReceivePolicyCustomizer = c -> c
+				.batchReceivePolicy(batchReceivePolicyOnBuilder);
+
+		MessagePayloadProcessor payloadProcessorOnBuilder = mock(MessagePayloadProcessor.class);
+		MessagePayloadProcessor payloadProcessorInProps = mock(MessagePayloadProcessor.class);
+		ConsumerBuilderCustomizer<String> payloadProcessorCustomizer = c -> c
+				.messagePayloadProcessor(payloadProcessorOnBuilder);
 
 		return Stream.of(arguments("loadConfNoDeadLetterPolicy", "deadLetterPolicy", null, null, null),
-				arguments("loadConfDeadLetterPolicyOnBuilder", "deadLetterPolicy", deadLetterPolicyOnBuilder, null,
+				arguments("loadConfDeadLetterPolicyOnBuilder", "deadLetterPolicy", deadLetterPolicyCustomizer, null,
 						deadLetterPolicyOnBuilder),
 				arguments("loadConfDeadLetterPolicyInProps", "deadLetterPolicy", null, deadLetterPolicyInProps,
 						deadLetterPolicyInProps),
-				arguments("loadConfDeadLetterPolicyOnBuilderAndInProps", "deadLetterPolicy", deadLetterPolicyOnBuilder,
+				arguments("loadConfDeadLetterPolicyOnBuilderAndInProps", "deadLetterPolicy", deadLetterPolicyCustomizer,
 						deadLetterPolicyInProps, deadLetterPolicyInProps),
+
+				arguments("loadConfNoMessageListener", "messageListener", null, null, null),
+				arguments("loadConfMessageListenerOnBuilder", "messageListener", messageListenerCustomizer, null,
+						messageListenerOnBuilder),
+				arguments("loadConfMessageListenerInProps", "messageListener", null, messageListenerInProps,
+						messageListenerInProps),
+				arguments("loadConfMessageListenerOnBuilderAndInProps", "messageListener", messageListenerCustomizer,
+						messageListenerInProps, messageListenerInProps),
+				arguments("loadConfNoConsumerEventListener", "consumerEventListener", null, null, null),
+				arguments("loadConfConsumerEventListenerOnBuilder", "consumerEventListener",
+						consumerEventListenerCustomizer, null, consumerEventListenerOnBuilder),
+				arguments("loadConfConsumerEventListenerInProps", "consumerEventListener", null,
+						consumerEventListenerInProps, consumerEventListenerInProps),
+				arguments("loadConfConsumerEventListenerOnBuilderAndInProps", "consumerEventListener",
+						consumerEventListenerCustomizer, consumerEventListenerInProps, consumerEventListenerInProps),
 				arguments("loadConfNoNegativeAckRedeliveryBackoff", "negativeAckRedeliveryBackoff", null, null, null),
 				arguments("loadConfNegativeAckRedeliveryBackoffOnBuilder", "negativeAckRedeliveryBackoff",
-						nackRedeliveryBackoffOnBuilder, null, nackRedeliveryBackoffOnBuilder),
+						nackRedeliveryBackoffCustomizer, null, nackRedeliveryBackoffOnBuilder),
 				arguments("loadConfNegativeAckRedeliveryBackoffInProps", "negativeAckRedeliveryBackoff", null,
 						nackRedeliveryBackoffInProps, nackRedeliveryBackoffInProps),
 				arguments("loadConfNegativeAckRedeliveryBackoffOnBuilderAndInProps", "negativeAckRedeliveryBackoff",
-						nackRedeliveryBackoffOnBuilder, nackRedeliveryBackoffInProps, nackRedeliveryBackoffInProps),
+						nackRedeliveryBackoffCustomizer, nackRedeliveryBackoffInProps, nackRedeliveryBackoffInProps),
 				arguments("loadConfNoAckRedeliveryBackoff", "ackTimeoutRedeliveryBackoff", null, null, null),
 				arguments("loadConfAckRedeliveryBackoffOnBuilder", "ackTimeoutRedeliveryBackoff",
-						ackRedeliveryBackoffOnBuilder, null, ackRedeliveryBackoffOnBuilder),
+						ackRedeliveryBackoffCustomizer, null, ackRedeliveryBackoffOnBuilder),
 				arguments("loadConfAckRedeliveryBackoffInProps", "ackTimeoutRedeliveryBackoff", null,
 						ackRedeliveryBackoffInProps, ackRedeliveryBackoffInProps),
 				arguments("loadConfAckRedeliveryBackoffOnBuilderAndInProps", "ackTimeoutRedeliveryBackoff",
-						ackRedeliveryBackoffOnBuilder, ackRedeliveryBackoffInProps, ackRedeliveryBackoffInProps));
+						ackRedeliveryBackoffCustomizer, ackRedeliveryBackoffInProps, ackRedeliveryBackoffInProps),
+				arguments("loadConfNoCryptoKeyReader", "cryptoKeyReader", null, null, null),
+				arguments("loadConfCryptoKeyReaderOnBuilder", "cryptoKeyReader", cryptoKeyReaderCustomizer, null,
+						cryptoKeyReaderOnBuilder),
+				arguments("loadConfCryptoKeyReaderInProps", "cryptoKeyReader", null, cryptoKeyReaderInProps,
+						cryptoKeyReaderInProps),
+				arguments("loadConfCryptoKeyReaderOnBuilderAndInProps", "cryptoKeyReader", cryptoKeyReaderCustomizer,
+						cryptoKeyReaderInProps, cryptoKeyReaderInProps),
+				arguments("loadConfNoMessageCrypto", "messageCrypto", null, null, null),
+				arguments("loadConfMessageCryptoOnBuilder", "messageCrypto", messageCryptoCustomizer, null,
+						messageCryptoOnBuilder),
+				arguments("loadConfMessageCryptoInProps", "messageCrypto", null, messageCryptoInProps,
+						messageCryptoInProps),
+				arguments("loadConfMessageCryptoOnBuilderAndInProps", "messageCrypto", messageCryptoCustomizer,
+						messageCryptoInProps, messageCryptoInProps),
+				arguments("loadConfNoBatchReceivePolicy", "batchReceivePolicy", null, null, null),
+				arguments("loadConfBatchReceivePolicyOnBuilder", "batchReceivePolicy", batchReceivePolicyCustomizer,
+						null, batchReceivePolicyOnBuilder),
+				arguments("loadConfBatchReceivePolicyInProps", "batchReceivePolicy", null, batchReceivePolicyInProps,
+						batchReceivePolicyInProps),
+				arguments("loadConfBatchReceivePolicyOnBuilderAndInProps", "batchReceivePolicy",
+						batchReceivePolicyCustomizer, batchReceivePolicyInProps, batchReceivePolicyInProps),
+				arguments("loadConfNoPayloadProcessor", "payloadProcessor", null, null, null),
+				arguments("loadConfPayloadProcessorOnBuilder", "payloadProcessor", payloadProcessorCustomizer, null,
+						payloadProcessorOnBuilder),
+				arguments("loadConfPayloadProcessorInProps", "payloadProcessor", null, payloadProcessorInProps,
+						payloadProcessorInProps),
+				arguments("loadConfPayloadProcessorOnBuilderAndInProps", "payloadProcessor", payloadProcessorCustomizer,
+						payloadProcessorInProps, payloadProcessorInProps));
 	}
 
 }

spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerTestUtils.java

@@ -16,9 +16,19 @@
 
 package org.springframework.pulsar.core;
 
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.spy;
 
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
 import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.PulsarClientException;
 
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.pulsar.listener.DefaultPulsarMessageListenerContainer;
@@ -35,11 +45,20 @@ public final class ConsumerTestUtils {
 	 * @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;
+	@SuppressWarnings("unchecked")
+	public static Consumer<String> startContainerAndSpyOnConsumer(
+			DefaultPulsarMessageListenerContainer<String> container)
+			throws PulsarClientException, ExecutionException, InterruptedException, TimeoutException {
+		CompletableFuture<Consumer<String>> consumerFuture = new CompletableFuture<>();
+		doAnswer(invocation -> {
+			Consumer<String> consumer = spy((Consumer<String>) invocation.callRealMethod());
+			consumerFuture.complete(consumer);
+			return consumer;
+		}).when(container.getPulsarConsumerFactory()).createConsumer(any(), isNull(), isNull(), anyList());
+
+		container.start();
+		Thread.sleep(1000);
+		return consumerFuture.get(1, TimeUnit.SECONDS);
 	}
 
 	/**

spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java

@@ -19,6 +19,7 @@ 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.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
@@ -170,8 +171,8 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
 
 		final PulsarClient pulsarClient = PulsarClient.builder()
 				.serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
-		final DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(
-				pulsarClient, config);
+		final DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(
+				new DefaultPulsarConsumerFactory<>(pulsarClient, config));
 		CountDownLatch latch = new CountDownLatch(10);
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
@@ -184,9 +185,8 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
 		pulsarContainerProperties.setSubscriptionType(SubscriptionType.Shared);
 		DefaultPulsarMessageListenerContainer<String> container = new DefaultPulsarMessageListenerContainer<>(
 				pulsarConsumerFactory, pulsarContainerProperties);
-		container.start();
 
-		final Consumer<?> containerConsumer = ConsumerTestUtils.spyOnConsumer(container);
+		final Consumer<String> containerConsumer = ConsumerTestUtils.startContainerAndSpyOnConsumer(container);
 
 		Map<String, Object> prodConfig = Collections.singletonMap("topicName", "dpmlct-015");
 		final DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(