Spring/spring-pulsar
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use type safe structures to configure the consumer

Browse Source
This commit is contained in:
Christophe Bornet
2022-11-03 18:43:07 +01:00
committed by Chris Bono
parent b4203bedc8
commit 5997a338bd
6 changed files with 308 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
spring-pulsar/src/main/java/org/springframework/pulsar/core/ConsumerBuilderCustomizer.java Normal file
View File

@@ -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 <T> The message payload type
* @author Christophe Bornet
*/
@FunctionalInterface
public interface ConsumerBuilderCustomizer<T> {
/**
* Customizes a {@link ConsumerBuilder}.
* @param consumerBuilder the builder to customize
*/
void customize(ConsumerBuilder<T> consumerBuilder);
}

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

@@ -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 <T> underlying payload type for the consumer.
* @author Soby Chacko
* @author Alexander Preuß
* @author Christophe Bornet
*/
public class DefaultPulsarConsumerFactory<T> implements PulsarConsumerFactory<T> {
private final Map<String, Object> consumerConfig = new HashMap<>();
private final Map<String, Object> consumerConfig;
private final List<Consumer<T>> consumers = new ArrayList<>();
private PulsarClient pulsarClient;
private final PulsarClient pulsarClient;
public DefaultPulsarConsumerFactory(PulsarClient pulsarClient) {
this(pulsarClient, Collections.emptyMap());
}
public DefaultPulsarConsumerFactory(PulsarClient pulsarClient, Map<String, Object> consumerConfig) {
this.pulsarClient = pulsarClient;
if (!CollectionUtils.isEmpty(consumerConfig)) {
this.consumerConfig.putAll(consumerConfig);
}
this.consumerConfig = Collections.unmodifiableMap(consumerConfig);
}
@Override
public Consumer<T> createConsumer(Schema<T> schema, Map<String, Object> propertiesToOverride)
public Consumer<T> createConsumer(Schema<T> schema) throws PulsarClientException {
return createConsumer(schema, null, null, Collections.emptyList());
}
@Override
public Consumer<T> createConsumer(Schema<T> schema, Collection<String> topics) throws PulsarClientException {
return createConsumer(schema, topics, null, Collections.emptyList());
}
@Override
public Consumer<T> createConsumer(Schema<T> schema, @Nullable Collection<String> topics,
@Nullable Map<String, String> properties, @Nullable List<ConsumerBuilderCustomizer<T>> customizers)
throws PulsarClientException {
ConsumerBuilder<T> consumerBuilder = this.pulsarClient.newConsumer(schema);
Map<String, Object> config = new HashMap<>(this.consumerConfig);
final ConsumerBuilder<T> consumerBuilder = this.pulsarClient.newConsumer(schema);
final Map<String, Object> 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<T> consumer = consumerBuilder.subscribe();
this.consumers.add(consumer);
return consumer;
}
@Override
public Consumer<T> createConsumer(Schema<T> schema, BatchReceivePolicy batchReceivePolicy,
Map<String, Object> propertiesToOverride) throws PulsarClientException {
final ConsumerBuilder<T> consumerBuilder = this.pulsarClient.newConsumer(schema);
final Map<String, Object> 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<T> consumer = consumerBuilder.subscribe();
this.consumers.add(consumer);
return consumer;

49
spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java
View File

@@ -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 <T> payload type for the consumer.
* @author Soby Chacko
* @author Christophe Bornet
*/
public interface PulsarConsumerFactory<T> {
Consumer<T> createConsumer(Schema<T> schema, Map<String, Object> 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<T> createConsumer(Schema<T> schema) throws PulsarClientException;
Consumer<T> createConsumer(Schema<T> schema, BatchReceivePolicy batchReceivePolicy,
Map<String, Object> 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<T> createConsumer(Schema<T> schema, Collection<String> 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<T> createConsumer(Schema<T> schema, @Nullable Collection<String> topics,
@Nullable Map<String, String> properties, @Nullable List<ConsumerBuilderCustomizer<T>> customizers)
throws PulsarClientException;
/**
* Return the configuration options to use when creating consumers.
* @return the configuration options
*/
Map<String, Object> getConsumerConfig();
}

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

@@ -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<T> extends AbstractPulsarMessageListenerContainer<T> {
@@ -236,12 +249,25 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
Map<String, Object> 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<String> topicNames = (Set<String>) propertiesToConsumer.remove("topicNames");
Map<String, String> properties = (Map<String, String>) propertiesToConsumer.remove("properties");
ConsumerBuilderCustomizer<T> 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<T> extends AbstractPulsarMess
}
}
/*
* Method to work around the issue that ConsumerBuilder::loadConf crashes if a
* deadLetterPolicy was set on the builder. To be removed once
* https://github.com/apache/pulsar/issues/11646 is fixed.
*/
@SuppressWarnings("unchecked")
private static <T> void loadConf(ConsumerBuilder<T> builder, Map<String, Object> properties) {
if (properties.containsKey("topicsPattern")) {
builder.topicsPattern(properties.get("topicsPattern").toString());
}
if (properties.containsKey("subscriptionName")) {
builder.subscriptionName(properties.get("subscriptionName").toString());
}
if (properties.containsKey("subscriptionType")) {
builder.subscriptionType(SubscriptionType.valueOf(properties.get("subscriptionType").toString()));
}
if (properties.containsKey("subscriptionMode")) {
builder.subscriptionMode(SubscriptionMode.valueOf(properties.get("subscriptionMode").toString()));
}
if (properties.containsKey("subscriptionProperties")) {
builder.subscriptionProperties((Map<String, String>) properties.get("subscriptionProperties"));
}
if (properties.containsKey("messageListener")) {
builder.messageListener((MessageListener<T>) properties.get("messageListener"));
}
if (properties.containsKey("consumerEventListener")) {
builder.consumerEventListener((ConsumerEventListener) properties.get("consumerEventListener"));
}
if (properties.containsKey("negativeAckRedeliveryBackoff")) {
builder.negativeAckRedeliveryBackoff(
(RedeliveryBackoff) properties.get("negativeAckRedeliveryBackoff"));
}
if (properties.containsKey("ackTimeoutRedeliveryBackoff")) {
builder.ackTimeoutRedeliveryBackoff((RedeliveryBackoff) properties.get("ackTimeoutRedeliveryBackoff"));
}
if (properties.containsKey("receiverQueueSize")) {
builder.receiverQueueSize(Integer.parseInt(properties.get("receiverQueueSize").toString()));
}
if (properties.containsKey("acknowledgementsGroupTimeMicros")) {
builder.acknowledgmentGroupTime(
Long.parseLong(properties.get("acknowledgementsGroupTimeMicros").toString()),
TimeUnit.MICROSECONDS);
}
if (properties.containsKey("negativeAckRedeliveryDelayMicros")) {
builder.negativeAckRedeliveryDelay(
Long.parseLong(properties.get("negativeAckRedeliveryDelayMicros").toString()),
TimeUnit.MICROSECONDS);
}
if (properties.containsKey("maxTotalReceiverQueueSizeAcrossPartitions")) {
builder.maxTotalReceiverQueueSizeAcrossPartitions(
Integer.parseInt(properties.get("maxTotalReceiverQueueSizeAcrossPartitions").toString()));
}
if (properties.containsKey("consumerName")) {
builder.consumerName(properties.get("consumerName").toString());
}
if (properties.containsKey("ackTimeoutMillis")) {
builder.ackTimeout(Long.parseLong(properties.get("ackTimeoutMillis").toString()),
TimeUnit.MILLISECONDS);
}
if (properties.containsKey("tickDurationMillis")) {
builder.ackTimeoutTickTime(Long.parseLong(properties.get("tickDurationMillis").toString()),
TimeUnit.MILLISECONDS);
}
if (properties.containsKey("priorityLevel")) {
builder.priorityLevel(Integer.parseInt(properties.get("priorityLevel").toString()));
}
if (properties.containsKey("maxPendingChunkedMessage")) {
builder.maxPendingChunkedMessage(
Integer.parseInt(properties.get("maxPendingChunkedMessage").toString()));
}
if (properties.containsKey("autoAckOldestChunkedMessageOnQueueFull")) {
builder.autoAckOldestChunkedMessageOnQueueFull(
Boolean.parseBoolean(properties.get("autoAckOldestChunkedMessageOnQueueFull").toString()));
}
if (properties.containsKey("expireTimeOfIncompleteChunkedMessageMillis")) {
builder.expireTimeOfIncompleteChunkedMessage(
Long.parseLong(properties.get("expireTimeOfIncompleteChunkedMessageMillis").toString()),
TimeUnit.MILLISECONDS);
}
if (properties.containsKey("cryptoKeyReader")) {
builder.cryptoKeyReader((CryptoKeyReader) properties.get("cryptoKeyReader"));
}
if (properties.containsKey("messageCrypto")) {
builder.messageCrypto((MessageCrypto) properties.get("messageCrypto"));
}
if (properties.containsKey("cryptoFailureAction")) {
builder.cryptoFailureAction(
ConsumerCryptoFailureAction.valueOf(properties.get("cryptoFailureAction").toString()));
}
if (properties.containsKey("readCompacted")) {
builder.readCompacted(Boolean.parseBoolean(properties.get("readCompacted").toString()));
}
if (properties.containsKey("subscriptionInitialPosition")) {
builder.subscriptionInitialPosition(
SubscriptionInitialPosition.valueOf(properties.get("subscriptionInitialPosition").toString()));
}
if (properties.containsKey("patternAutoDiscoveryPeriod")) {
builder.patternAutoDiscoveryPeriod(
Integer.parseInt(properties.get("patternAutoDiscoveryPeriod").toString()), TimeUnit.SECONDS);
}
if (properties.containsKey("regexSubscriptionMode")) {
builder.subscriptionTopicsMode(
RegexSubscriptionMode.valueOf(properties.get("regexSubscriptionMode").toString()));
}
if (properties.containsKey("deadLetterPolicy")) {
builder.deadLetterPolicy((DeadLetterPolicy) properties.get("deadLetterPolicy"));
}
if (properties.containsKey("retryEnable")) {
builder.enableRetry(Boolean.parseBoolean(properties.get("retryEnable").toString()));
}
if (properties.containsKey("batchReceivePolicy")) {
builder.batchReceivePolicy((BatchReceivePolicy) properties.get("batchReceivePolicy"));
}
if (properties.containsKey("autoUpdatePartitions")) {
builder.autoUpdatePartitions(Boolean.parseBoolean(properties.get("autoUpdatePartitions").toString()));
}
if (properties.containsKey("autoUpdatePartitionsIntervalSeconds")) {
builder.autoUpdatePartitionsInterval(
Integer.parseInt(properties.get("autoUpdatePartitionsIntervalSeconds").toString()),
TimeUnit.SECONDS);
}
if (properties.containsKey("replicateSubscriptionState")) {
builder.replicateSubscriptionState(
Boolean.parseBoolean(properties.get("replicateSubscriptionState").toString()));
}
if (properties.containsKey("resetIncludeHead")) {
builder.startMessageIdInclusive();
}
if (properties.containsKey("keySharedPolicy")) {
builder.keySharedPolicy((KeySharedPolicy) properties.get("keySharedPolicy"));
}
if (properties.containsKey("batchIndexAckEnabled")) {
builder.enableBatchIndexAcknowledgment(
Boolean.parseBoolean(properties.get("batchIndexAckEnabled").toString()));
}
if (properties.containsKey("ackReceiptEnabled")) {
builder.isAckReceiptEnabled(Boolean.parseBoolean(properties.get("ackReceiptEnabled").toString()));
}
if (properties.containsKey("poolMessages")) {
builder.poolMessages(Boolean.parseBoolean(properties.get("poolMessages").toString()));
}
if (properties.containsKey("payloadProcessor")) {
builder.messagePayloadProcessor((MessagePayloadProcessor) properties.get("payloadProcessor"));
}
if (properties.containsKey("startPaused")) {
builder.startPaused(Boolean.parseBoolean(properties.get("startPaused").toString()));
}
}
private Map<String, Object> extractDirectConsumerProperties() {
Properties propertyOverrides = this.containerProperties.getPulsarConsumerProperties();
return propertyOverrides.entrySet().stream().collect(Collectors.toMap(e -> String.valueOf(e.getKey()),

9
spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java
View File

@@ -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<String> consumerFactory = mock(PulsarConsumerFactory.class);
Consumer<String> 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();

6
spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java
View File

@@ -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);