diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/ConsumerBuilderCustomizer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/ConsumerBuilderCustomizer.java new file mode 100644 index 00000000..7baa51e1 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/ConsumerBuilderCustomizer.java @@ -0,0 +1,36 @@ +/* + * 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 org.apache.pulsar.client.api.ConsumerBuilder; + +/** + * The interface to customize a {@link ConsumerBuilder}. + * + * @param The message payload type + * @author Christophe Bornet + */ +@FunctionalInterface +public interface ConsumerBuilderCustomizer { + + /** + * Customizes a {@link ConsumerBuilder}. + * @param consumerBuilder the builder to customize + */ + void customize(ConsumerBuilder consumerBuilder); + +} 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 7a2a97c6..455205de 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 @@ -17,11 +17,14 @@ package org.springframework.pulsar.core; import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.TreeMap; -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.DeadLetterPolicy; @@ -30,6 +33,7 @@ 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; import org.springframework.util.CollectionUtils; /** @@ -38,75 +42,78 @@ import org.springframework.util.CollectionUtils; * @param underlying payload type for the consumer. * @author Soby Chacko * @author Alexander Preuß + * @author Christophe Bornet */ public class DefaultPulsarConsumerFactory implements PulsarConsumerFactory { - private final Map consumerConfig = new HashMap<>(); + private final Map consumerConfig; private final List> consumers = new ArrayList<>(); - private PulsarClient pulsarClient; + private final PulsarClient pulsarClient; + + public DefaultPulsarConsumerFactory(PulsarClient pulsarClient) { + this(pulsarClient, Collections.emptyMap()); + } public DefaultPulsarConsumerFactory(PulsarClient pulsarClient, Map consumerConfig) { this.pulsarClient = pulsarClient; - if (!CollectionUtils.isEmpty(consumerConfig)) { - this.consumerConfig.putAll(consumerConfig); - } + this.consumerConfig = Collections.unmodifiableMap(consumerConfig); } @Override - public Consumer createConsumer(Schema schema, Map propertiesToOverride) + public Consumer createConsumer(Schema schema) throws PulsarClientException { + return createConsumer(schema, null, null, Collections.emptyList()); + } + + @Override + public Consumer createConsumer(Schema schema, Collection topics) throws PulsarClientException { + return createConsumer(schema, topics, null, Collections.emptyList()); + } + + @Override + public Consumer createConsumer(Schema schema, @Nullable Collection topics, + @Nullable Map properties, @Nullable List> customizers) throws PulsarClientException { + ConsumerBuilder consumerBuilder = this.pulsarClient.newConsumer(schema); + Map config = new HashMap<>(this.consumerConfig); - final ConsumerBuilder consumerBuilder = this.pulsarClient.newConsumer(schema); - - final Map properties = new HashMap<>(this.consumerConfig); - properties.putAll(propertiesToOverride); - - if (!CollectionUtils.isEmpty(properties)) { - consumerBuilder.loadConf(properties); + if (topics != null) { + config.put("topicNames", new HashSet<>(topics)); + } + if (properties != null) { + config.put("properties", new TreeMap<>(properties)); } - Consumer consumer = consumerBuilder.subscribe(); - this.consumers.add(consumer); - return consumer; - } - - @Override - public Consumer createConsumer(Schema schema, BatchReceivePolicy batchReceivePolicy, - Map propertiesToOverride) throws PulsarClientException { - - final ConsumerBuilder consumerBuilder = this.pulsarClient.newConsumer(schema); - final Map properties = new HashMap<>(this.consumerConfig); - properties.putAll(propertiesToOverride); // 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 (properties.containsKey("deadLetterPolicy")) { - deadLetterPolicy = (DeadLetterPolicy) properties.remove("deadLetterPolicy"); + if (config.containsKey("deadLetterPolicy")) { + deadLetterPolicy = (DeadLetterPolicy) config.remove("deadLetterPolicy"); } - if (!CollectionUtils.isEmpty(properties)) { - consumerBuilder.loadConf(properties); - } + consumerBuilder.loadConf(config); if (deadLetterPolicy != null) { consumerBuilder.deadLetterPolicy(deadLetterPolicy); } - if (properties.containsKey("negativeAckRedeliveryBackoff")) { - final RedeliveryBackoff negativeAckRedeliveryBackoff = (RedeliveryBackoff) properties + if (config.containsKey("negativeAckRedeliveryBackoff")) { + RedeliveryBackoff negativeAckRedeliveryBackoff = (RedeliveryBackoff) config .get("negativeAckRedeliveryBackoff"); consumerBuilder.negativeAckRedeliveryBackoff(negativeAckRedeliveryBackoff); } - if (properties.containsKey("ackTimeoutRedeliveryBackoff")) { - final RedeliveryBackoff ackTimeoutRedeliveryBackoff = (RedeliveryBackoff) properties + if (config.containsKey("ackTimeoutRedeliveryBackoff")) { + RedeliveryBackoff ackTimeoutRedeliveryBackoff = (RedeliveryBackoff) config .get("ackTimeoutRedeliveryBackoff"); consumerBuilder.ackTimeoutRedeliveryBackoff(ackTimeoutRedeliveryBackoff); } - consumerBuilder.batchReceivePolicy(batchReceivePolicy); + if (!CollectionUtils.isEmpty(customizers)) { + customizers.forEach(customizer -> customizer.customize(consumerBuilder)); + } + Consumer consumer = consumerBuilder.subscribe(); this.consumers.add(consumer); return consumer; 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 ca378aaf..e8967737 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 @@ -16,26 +16,67 @@ package org.springframework.pulsar.core; +import java.util.Collection; +import java.util.List; import java.util.Map; -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.PulsarClientException; import org.apache.pulsar.client.api.Schema; +import org.springframework.lang.Nullable; + /** * Pulsar consumer factory interface. * * @param payload type for the consumer. * @author Soby Chacko + * @author Christophe Bornet */ public interface PulsarConsumerFactory { - Consumer createConsumer(Schema schema, Map propertiesToOverride) throws PulsarClientException; + /** + * Create a consumer. + * @param schema the schema of the messages to be sent + * @return the consumer + * @throws PulsarClientException if any error occurs + */ + Consumer createConsumer(Schema schema) throws PulsarClientException; - Consumer createConsumer(Schema schema, BatchReceivePolicy batchReceivePolicy, - Map propertiesToOverride) throws PulsarClientException; + /** + * Create a consumer. + * @param schema the schema of the messages to be sent + * @param topics the topics the consumer will subscribe to + * @return the consumer + * @throws PulsarClientException if any error occurs + */ + Consumer createConsumer(Schema schema, Collection topics) throws PulsarClientException; + /** + * Create a consumer. + * @param schema the schema of the messages to be sent + * @param topics the topics the consumer will subscribe to overriding the default ones + * 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 customizers the optional list of customizers to apply to the consumer + * builder + * @return the consumer + * @throws PulsarClientException if any error occurs + */ + Consumer createConsumer(Schema schema, @Nullable Collection topics, + @Nullable Map properties, @Nullable List> customizers) + throws PulsarClientException; + + /** + * Return the configuration options to use when creating consumers. + * @return the configuration options + */ Map getConsumerConfig(); } 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 08190e02..56d8e656 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 @@ -18,6 +18,7 @@ package org.springframework.pulsar.listener; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.List; @@ -36,14 +37,24 @@ 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; @@ -51,6 +62,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.ConsumerBuilderCustomizer; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.event.ConsumerFailedToStartEvent; import org.springframework.pulsar.event.ConsumerStartedEvent; @@ -72,6 +84,7 @@ import io.micrometer.observation.ObservationRegistry; * @author Soby Chacko * @author Alexander Preuß * @author Chris Bono + * @author Christophe Bornet */ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMessageListenerContainer { @@ -236,12 +249,25 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess Map propertiesToConsumer = extractDirectConsumerProperties(); populateAllNecessaryPropertiesIfNeedBe(propertiesToConsumer); - final BatchReceivePolicy batchReceivePolicy = new BatchReceivePolicy.Builder() + BatchReceivePolicy batchReceivePolicy = new BatchReceivePolicy.Builder() .maxNumMessages(containerProperties.getMaxNumMessages()) .maxNumBytes(containerProperties.getMaxNumBytes()) .timeout(containerProperties.getBatchTimeoutMillis(), TimeUnit.MILLISECONDS).build(); + + /* + * topicNames and properties must not be added through the builder + * customizer as ConsumerBuilder::topics and ConsumerBuilder::properties + * don't replace but add to the existing topics/properties. + */ + Set topicNames = (Set) propertiesToConsumer.remove("topicNames"); + Map properties = (Map) propertiesToConsumer.remove("properties"); + + ConsumerBuilderCustomizer customizer = builder -> { + loadConf(builder, propertiesToConsumer); + builder.batchReceivePolicy(batchReceivePolicy); + }; this.consumer = getPulsarConsumerFactory().createConsumer((Schema) containerProperties.getSchema(), - batchReceivePolicy, propertiesToConsumer); + topicNames, properties, Collections.singletonList(customizer)); Assert.state(this.consumer != null, "Unable to create a consumer"); } catch (PulsarClientException e) { @@ -249,6 +275,155 @@ 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/listener/ConcurrentPulsarMessageListenerContainerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java index 02ce951b..924beb29 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java @@ -20,15 +20,15 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.awaitility.Awaitility.await; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.time.Duration; -import java.util.Map; -import org.apache.pulsar.client.api.BatchReceivePolicy; import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.DeadLetterPolicy; import org.apache.pulsar.client.api.Messages; @@ -157,7 +157,7 @@ public class ConcurrentPulsarMessageListenerContainerTests { concurrentContainer.start(); await().atMost(Duration.ofSeconds(10)).untilAsserted(() -> verify(pulsarConsumerFactory, times(3)) - .createConsumer(any(Schema.class), any(BatchReceivePolicy.class), any(Map.class))); + .createConsumer(any(Schema.class), isNull(), isNull(), anyList())); await().atMost(Duration.ofSeconds(10)).untilAsserted(() -> verify(consumer, times(3)).batchReceive()); } @@ -177,8 +177,7 @@ public class ConcurrentPulsarMessageListenerContainerTests { throws Exception { PulsarConsumerFactory consumerFactory = mock(PulsarConsumerFactory.class); Consumer consumer = mock(Consumer.class); - when(consumerFactory.createConsumer(any(Schema.class), any(BatchReceivePolicy.class), any(Map.class))) - .thenReturn(consumer); + when(consumerFactory.createConsumer(any(Schema.class), isNull(), isNull(), anyList())).thenReturn(consumer); when(consumer.batchReceive()).thenReturn(mock(Messages.class)); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java index 5262aa75..2b29ecec 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java @@ -299,7 +299,7 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { void pulsarListenerWithAckTimeoutRedeliveryBackoff(@Autowired PulsarListenerEndpointRegistry registry) throws Exception { pulsarTemplate.send("withAckTimeoutRedeliveryBackoff-test-topic", "hello john doe"); - assertThat(ackTimeoutRedeliveryBackoffLatch.await(15, TimeUnit.SECONDS)).isTrue(); + assertThat(ackTimeoutRedeliveryBackoffLatch.await(60, TimeUnit.SECONDS)).isTrue(); } @EnablePulsar @@ -310,7 +310,7 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { subscriptionName = "withAckTimeoutRedeliveryBackoffSubscription", topics = "withAckTimeoutRedeliveryBackoff-test-topic", ackTimeoutRedeliveryBackoff = "ackTimeoutRedeliveryBackoff", - subscriptionType = SubscriptionType.Shared, properties = { "ackTimeoutMillis=1" }) + subscriptionType = SubscriptionType.Shared, properties = { "ackTimeoutMillis=1000" }) void listen(String msg) { ackTimeoutRedeliveryBackoffLatch.countDown(); throw new RuntimeException(); @@ -390,7 +390,7 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { @PulsarListener(id = "deadLetterPolicyListener", subscriptionName = "deadLetterPolicySubscription", topics = "dlpt-topic-1", deadLetterPolicy = "deadLetterPolicy", - subscriptionType = SubscriptionType.Shared, properties = { "ackTimeoutMillis=1" }) + subscriptionType = SubscriptionType.Shared, properties = { "ackTimeoutMillis=1000" }) void listen(String msg) { latch.countDown(); throw new RuntimeException("fail " + msg);