diff --git a/spring-pulsar-dependencies/build.gradle b/spring-pulsar-dependencies/build.gradle index e4c56923..4affb96f 100644 --- a/spring-pulsar-dependencies/build.gradle +++ b/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' } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/ConsumerBuilderConfigurationUtil.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/ConsumerBuilderConfigurationUtil.java index 105a7f16..b9221283 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/ConsumerBuilderConfigurationUtil.java +++ b/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}. *

* The main purpose is to work around the underlying - * Pulsar issue where - * {@code ConsumerBuilder::loadConf} sets {@code @JsonIgnore} fields to null and crashes - * if a {@code deadLetterPolicy} was set on the builder. + * Pulsar issue where + * {@link ConsumerBuilder#loadConf} sets {@code @JsonIgnore} fields to null. *

* 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 Pulsar issue. + * loads non-serializable properties. See + * Pulsar PR. * @param builder the builder * @param properties the properties to set on the builder * @param the payload type */ - @SuppressWarnings("unchecked") public static void loadConf(ConsumerBuilder builder, Map properties) { ConsumerConfigurationData builderConf = ((ConsumerBuilderImpl) builder).getConf(); Map 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 getValueToApplyToBuilderAfterLoadConf(Supplier builderGetter, Consumer builderSetter, - Map properties, String propertyName) { - T value = (T) properties.getOrDefault(propertyName, builderGetter.get()); - - if ("deadLetterPolicy".equals(propertyName)) { - builderSetter.accept(null); - properties.remove(propertyName); + private static void applyValueToBuilderAfterLoadConf(Supplier confGetter, Consumer builderSetter, + Map properties, String key) { + T value = (T) properties.getOrDefault(key, confGetter.get()); + if (value != null) { + builderSetter.accept(value); } - return value; } } 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 5e403930..39a2e2a7 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 @@ -84,8 +84,6 @@ public class DefaultPulsarConsumerFactory implements PulsarConsumerFactory 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)) { 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 38e9f36c..0f430c9e 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 @@ -254,8 +254,6 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess Map properties = (Map) propertiesToConsumer.remove("properties"); ConsumerBuilderCustomizer customizer = builder -> { - // Replace w/ consumerBuilder.loadConf after - // https://github.com/apache/pulsar/issues/11646 ConsumerBuilderConfigurationUtil.loadConf(builder, propertiesToConsumer); builder.batchReceivePolicy(batchReceivePolicy); }; 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 f6c15582..284a4f70 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,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 pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>( - pulsarClient, config); + final DefaultPulsarConsumerFactory 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 container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); - container.start(); - final Consumer containerConsumer = ConsumerTestUtils.spyOnConsumer(container); + final Consumer 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 pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>( - pulsarClient, config); + final DefaultPulsarConsumerFactory 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 container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); - container.start(); - final Consumer containerConsumer = ConsumerTestUtils.spyOnConsumer(container); + final Consumer containerConsumer = ConsumerTestUtils.startContainerAndSpyOnConsumer(container); Map 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 pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>( - pulsarClient, config); + final DefaultPulsarConsumerFactory 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 container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); - container.start(); - final Consumer containerConsumer = ConsumerTestUtils.spyOnConsumer(container); + final Consumer 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 pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>( - pulsarClient, config); + final DefaultPulsarConsumerFactory pulsarConsumerFactory = spy( + new DefaultPulsarConsumerFactory<>(pulsarClient, config)); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); final List acksObjects = new ArrayList<>(); @@ -233,8 +231,7 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { pulsarContainerProperties.setAckMode(AckMode.MANUAL); DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); - container.start(); - final Consumer containerConsumer = ConsumerTestUtils.spyOnConsumer(container); + final Consumer 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 pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>( - pulsarClient, config); + final DefaultPulsarConsumerFactory 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 container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); - container.start(); - final Consumer containerConsumer = ConsumerTestUtils.spyOnConsumer(container); + final Consumer containerConsumer = ConsumerTestUtils.startContainerAndSpyOnConsumer(container); Map 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 pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>( - pulsarClient, config); + final DefaultPulsarConsumerFactory 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 container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); - container.start(); - final Consumer containerConsumer = ConsumerTestUtils.spyOnConsumer(container); + final Consumer containerConsumer = ConsumerTestUtils.startContainerAndSpyOnConsumer(container); Map prodConfig = new HashMap<>(); prodConfig.put("topicName", "cons-ack-tests-016"); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerBuilderConfigurationUtilTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerBuilderConfigurationUtilTests.java index e7d446f3..5aebe43f 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerBuilderConfigurationUtilTests.java +++ b/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 objOnBuilder, + @Nullable Object objOnProps, @Nullable Object expectedObj) { if (objOnBuilder != null) { - ReflectionTestUtils.invokeSetterMethod(builder, propName, objOnBuilder); + objOnBuilder.customize(builder); } Map props = new HashMap<>(); @@ -79,44 +84,116 @@ public class ConsumerBuilderConfigurationUtilTests { assertThat(props).isEqualTo(propsBeforeUtil); } + @SuppressWarnings({ "unchecked", "rawtypes" }) private static Stream loadConfTestProvider() { - DeadLetterPolicy deadLetterPolicyOnBuilder = DeadLetterPolicy.builder().deadLetterTopic("dlt-topic") .maxRedeliverCount(1).build(); DeadLetterPolicy deadLetterPolicyInProps = DeadLetterPolicy.builder().deadLetterTopic("dlt-topic") .maxRedeliverCount(2).build(); + ConsumerBuilderCustomizer deadLetterPolicyCustomizer = c -> c + .deadLetterPolicy(deadLetterPolicyOnBuilder); - RedeliveryBackoff nackRedeliveryBackoffOnBuilder = MultiplierRedeliveryBackoff.builder().minDelayMs(1000) - .maxDelayMs(5000).build(); - RedeliveryBackoff nackRedeliveryBackoffInProps = MultiplierRedeliveryBackoff.builder().minDelayMs(2000) - .maxDelayMs(4000).build(); + MessageListener messageListenerOnBuilder = mock(MessageListener.class); + MessageListener messageListenerInProps = mock(MessageListener.class); + ConsumerBuilderCustomizer 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 consumerEventListenerCustomizer = c -> c + .consumerEventListener(consumerEventListenerOnBuilder); + + RedeliveryBackoff nackRedeliveryBackoffOnBuilder = mock(RedeliveryBackoff.class); + RedeliveryBackoff nackRedeliveryBackoffInProps = mock(RedeliveryBackoff.class); + ConsumerBuilderCustomizer nackRedeliveryBackoffCustomizer = c -> c + .negativeAckRedeliveryBackoff(nackRedeliveryBackoffOnBuilder); + + RedeliveryBackoff ackRedeliveryBackoffOnBuilder = mock(RedeliveryBackoff.class); + RedeliveryBackoff ackRedeliveryBackoffInProps = mock(RedeliveryBackoff.class); + ConsumerBuilderCustomizer ackRedeliveryBackoffCustomizer = c -> c + .ackTimeoutRedeliveryBackoff(ackRedeliveryBackoffOnBuilder); + + CryptoKeyReader cryptoKeyReaderOnBuilder = mock(CryptoKeyReader.class); + CryptoKeyReader cryptoKeyReaderInProps = mock(CryptoKeyReader.class); + ConsumerBuilderCustomizer cryptoKeyReaderCustomizer = c -> c.cryptoKeyReader(cryptoKeyReaderOnBuilder); + + MessageCrypto messageCryptoOnBuilder = mock(MessageCrypto.class); + MessageCrypto messageCryptoInProps = mock(MessageCrypto.class); + ConsumerBuilderCustomizer messageCryptoCustomizer = c -> c.messageCrypto(messageCryptoOnBuilder); + + BatchReceivePolicy batchReceivePolicyOnBuilder = mock(BatchReceivePolicy.class); + BatchReceivePolicy batchReceivePolicyInProps = mock(BatchReceivePolicy.class); + ConsumerBuilderCustomizer batchReceivePolicyCustomizer = c -> c + .batchReceivePolicy(batchReceivePolicyOnBuilder); + + MessagePayloadProcessor payloadProcessorOnBuilder = mock(MessagePayloadProcessor.class); + MessagePayloadProcessor payloadProcessorInProps = mock(MessagePayloadProcessor.class); + ConsumerBuilderCustomizer 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)); } } 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 index 610e5fb0..66c52dbe 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerTestUtils.java +++ b/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 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 startContainerAndSpyOnConsumer( + DefaultPulsarMessageListenerContainer container) + throws PulsarClientException, ExecutionException, InterruptedException, TimeoutException { + CompletableFuture> consumerFuture = new CompletableFuture<>(); + doAnswer(invocation -> { + Consumer consumer = spy((Consumer) 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); } /** 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 303fde7d..5ed5eed2 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 @@ -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 pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>( - pulsarClient, config); + final DefaultPulsarConsumerFactory 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 container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); - container.start(); - final Consumer containerConsumer = ConsumerTestUtils.spyOnConsumer(container); + final Consumer containerConsumer = ConsumerTestUtils.startContainerAndSpyOnConsumer(container); Map prodConfig = Collections.singletonMap("topicName", "dpmlct-015"); final DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(