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 new file mode 100644 index 00000000..105a7f16 --- /dev/null +++ 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}. + *

+ * 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. + *

+ * 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 Pulsar issue. + * @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); + + // 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 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); + } + 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 455205de..5e403930 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 @@ -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 implements PulsarConsumerFactory } @Override - public Consumer createConsumer(Schema schema, Collection topics) throws PulsarClientException { + public Consumer createConsumer(Schema schema, @Nullable Collection topics) + throws PulsarClientException { return createConsumer(schema, topics, null, Collections.emptyList()); } @@ -85,30 +84,9 @@ public class DefaultPulsarConsumerFactory implements PulsarConsumerFactory 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)); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java index e8967737..08528a56 100644 --- 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 payload type for the consumer. * @author Soby Chacko * @author Christophe Bornet + * @author Chris Bono */ public interface PulsarConsumerFactory { @@ -47,11 +48,12 @@ public interface PulsarConsumerFactory { /** * 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 createConsumer(Schema schema, Collection topics) throws PulsarClientException; + Consumer createConsumer(Schema schema, @Nullable Collection topics) throws PulsarClientException; /** * Create a consumer. @@ -60,10 +62,11 @@ public interface PulsarConsumerFactory { * 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 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 56d8e656..38e9f36c 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java @@ -37,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 extends AbstractPulsarMess Map properties = (Map) propertiesToConsumer.remove("properties"); ConsumerBuilderCustomizer 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 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 void loadConf(ConsumerBuilder builder, Map 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) properties.get("subscriptionProperties")); - } - if (properties.containsKey("messageListener")) { - builder.messageListener((MessageListener) 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 extractDirectConsumerProperties() { Properties propertyOverrides = this.containerProperties.getPulsarConsumerProperties(); return propertyOverrides.entrySet().stream().collect(Collectors.toMap(e -> String.valueOf(e.getKey()), 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 new file mode 100644 index 00000000..e7d446f3 --- /dev/null +++ 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 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 props = new HashMap<>(); + if (objOnProps != null) { + props.put(propName, objOnProps); + } + Map 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 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)); + } + +} 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 05683425..303fde7d 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 @@ -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 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 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 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 dlqContainer = new DefaultPulsarMessageListenerContainer<>( + pulsarConsumerFactory, dlqContainerProperties); + dlqContainer.start(); + + PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); + pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener) (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 container = new DefaultPulsarMessageListenerContainer<>( + pulsarConsumerFactory, pulsarContainerProperties); + container.start(); + + Map prodConfig = Collections.singletonMap("topicName", "dpmlct-016"); + final DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>( + pulsarClient, prodConfig); + final PulsarTemplate 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(); + } + }