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

Add DeadLetterPolicy to PulsarListener

Browse Source 
Closes #79
This commit is contained in:
Alexander Preuß
2022-09-06 10:48:53 +02:00
committed by Chris Bono
parent 78efde7cd6
commit 25dcfab607
11 changed files with 213 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java
View File

@@ -46,6 +46,7 @@ import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
*
* @author Soby Chacko
* @author Chris Bono
* @author Alexander Preuß
*/
@Target({ ElementType.TYPE, ElementType.METHOD, ElementType.ANNOTATION_TYPE })
@Retention(RetentionPolicy.RUNTIME)
@@ -166,8 +167,16 @@ public @interface PulsarListener {
* The bean name or a 'SpEL' expression that resolves to a
* {@link org.apache.pulsar.client.api.RedeliveryBackoff} to use on the consumer to
* control the redelivery backoff of messages after a negative ack.
* @return the bean name or empty string to not set the backoff
* @return the bean name or empty string to not set the backoff.
*/
String negativeAckRedeliveryBackoff() default "";
/**
* The bean name or a 'SpEL' expression that resolves to a
* {@link org.apache.pulsar.client.api.DeadLetterPolicy} to use on the consumer to
* configure a dead letter policy for message redelivery.
* @return the bean name or empty string to not set any dead letter policy.
*/
String deadLetterPolicy() default "";
}

17
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java
View File

@@ -39,6 +39,7 @@ import java.util.function.BiFunction;
import java.util.stream.Collectors;
import org.apache.commons.logging.LogFactory;
import org.apache.pulsar.client.api.DeadLetterPolicy;
import org.apache.pulsar.client.api.RedeliveryBackoff;
import org.apache.pulsar.client.api.SubscriptionType;
@@ -363,6 +364,7 @@ public class PulsarListenerAnnotationBeanPostProcessor<K, V>
endpoint.setBeanFactory(this.beanFactory);
resolveNegativeAckRedeliveryBackoff(endpoint, pulsarListener);
resolveDeadLetterPolicy(endpoint, pulsarListener);
}
private void resolveNegativeAckRedeliveryBackoff(MethodPulsarListenerEndpoint<?> endpoint,
@@ -381,6 +383,21 @@ public class PulsarListenerAnnotationBeanPostProcessor<K, V>
}
}
private void resolveDeadLetterPolicy(MethodPulsarListenerEndpoint<?> endpoint, PulsarListener pulsarListener) {
Object deadLetterPolicy = resolveExpression(pulsarListener.deadLetterPolicy());
if (deadLetterPolicy instanceof DeadLetterPolicy) {
endpoint.setDeadLetterPolicy((DeadLetterPolicy) deadLetterPolicy);
}
else {
String deadLetterPolicyBeanName = resolveExpressionAsString(pulsarListener.deadLetterPolicy(),
"deadLetterPolicy");
if (StringUtils.hasText(deadLetterPolicyBeanName)) {
endpoint.setDeadLetterPolicy(
this.beanFactory.getBean(deadLetterPolicyBeanName, DeadLetterPolicy.class));
}
}
}
private Integer resolveExpressionAsInteger(String value, String attribute) {
Object resolved = resolveExpression(value);
Integer result = null;

8
spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java
View File

@@ -24,6 +24,7 @@ import java.util.function.Function;
import org.apache.commons.logging.LogFactory;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.DeadLetterPolicy;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.Messages;
import org.apache.pulsar.client.api.RedeliveryBackoff;
@@ -80,6 +81,8 @@ public class MethodPulsarListenerEndpoint<V> extends AbstractPulsarListenerEndpo
private RedeliveryBackoff negativeAckRedeliveryBackoff;
private DeadLetterPolicy deadLetterPolicy;
public void setBean(Object bean) {
this.bean = bean;
}
@@ -185,6 +188,7 @@ public class MethodPulsarListenerEndpoint<V> extends AbstractPulsarListenerEndpo
pulsarContainerProperties.setSchemaType(type);
container.setNegativeAckRedeliveryBackoff(this.negativeAckRedeliveryBackoff);
container.setDeadLetterPolicy(this.deadLetterPolicy);
return messageListener;
}
@@ -263,4 +267,8 @@ public class MethodPulsarListenerEndpoint<V> extends AbstractPulsarListenerEndpo
this.negativeAckRedeliveryBackoff = negativeAckRedeliveryBackoff;
}
public void setDeadLetterPolicy(DeadLetterPolicy deadLetterPolicy) {
this.deadLetterPolicy = deadLetterPolicy;
}
}

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

@@ -24,6 +24,7 @@ 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.DeadLetterPolicy;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.RedeliveryBackoff;
@@ -36,6 +37,7 @@ import org.springframework.util.CollectionUtils;
*
* @param <T> underlying payload type for the consumer.
* @author Soby Chacko
* @author Alexander Preuß
*/
public class DefaultPulsarConsumerFactory<T> implements PulsarConsumerFactory<T> {
@@ -77,10 +79,21 @@ public class DefaultPulsarConsumerFactory<T> implements PulsarConsumerFactory<T>
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 (!CollectionUtils.isEmpty(properties)) {
consumerBuilder.loadConf(properties);
}
if (deadLetterPolicy != null) {
consumerBuilder.deadLetterPolicy(deadLetterPolicy);
}
if (properties.containsKey("negativeAckRedeliveryBackoff")) {
final RedeliveryBackoff negativeAckRedeliveryBackoff = (RedeliveryBackoff) properties
.get("negativeAckRedeliveryBackoff");

13
spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java
View File

@@ -17,6 +17,7 @@
package org.springframework.pulsar.listener;
import org.apache.commons.logging.LogFactory;
import org.apache.pulsar.client.api.DeadLetterPolicy;
import org.apache.pulsar.client.api.RedeliveryBackoff;
import org.springframework.beans.BeansException;
@@ -35,6 +36,7 @@ import org.springframework.util.Assert;
*
* @param <T> message type.
* @author Soby Chacko
* @author Alexander Preuß
*/
public abstract class AbstractPulsarMessageListenerContainer<T> implements PulsarMessageListenerContainer,
BeanNameAware, ApplicationEventPublisherAware, ApplicationContextAware {
@@ -61,6 +63,8 @@ public abstract class AbstractPulsarMessageListenerContainer<T> implements Pulsa
protected RedeliveryBackoff negativeAckRedeliveryBackoff;
protected DeadLetterPolicy deadLetterPolicy;
@SuppressWarnings("unchecked")
protected AbstractPulsarMessageListenerContainer(PulsarConsumerFactory<? super T> pulsarConsumerFactory,
PulsarContainerProperties pulsarContainerProperties) {
@@ -185,4 +189,13 @@ public abstract class AbstractPulsarMessageListenerContainer<T> implements Pulsa
return this.negativeAckRedeliveryBackoff;
}
@Override
public void setDeadLetterPolicy(DeadLetterPolicy deadLetterPolicy) {
this.deadLetterPolicy = deadLetterPolicy;
}
public DeadLetterPolicy getDeadLetterPolicy() {
return this.deadLetterPolicy;
}
}

2
spring-pulsar/src/main/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainer.java
View File

@@ -35,6 +35,7 @@ import org.springframework.util.Assert;
*
* @param <T> the payload type.
* @author Soby Chacko
* @author Alexander Preuß
*/
public class ConcurrentPulsarMessageListenerContainer<T> extends AbstractPulsarMessageListenerContainer<T> {
@@ -109,6 +110,7 @@ public class ConcurrentPulsarMessageListenerContainer<T> extends AbstractPulsarM
container.getContainerProperties().setConsumerTaskExecutor(exec);
}
container.setNegativeAckRedeliveryBackoff(this.negativeAckRedeliveryBackoff);
container.setDeadLetterPolicy(this.deadLetterPolicy);
}
@Override

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

@@ -32,6 +32,7 @@ import java.util.stream.StreamSupport;
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.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.MessageListener;
@@ -239,6 +240,10 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
if (negativeAckRedeliveryBackoff != null) {
currentProperties.put("negativeAckRedeliveryBackoff", negativeAckRedeliveryBackoff);
}
final DeadLetterPolicy deadLetterPolicy = DefaultPulsarMessageListenerContainer.this.deadLetterPolicy;
if (deadLetterPolicy != null) {
currentProperties.put("deadLetterPolicy", deadLetterPolicy);
}
}
@Override

3
spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageListenerContainer.java
View File

@@ -16,6 +16,7 @@
package org.springframework.pulsar.listener;
import org.apache.pulsar.client.api.DeadLetterPolicy;
import org.apache.pulsar.client.api.RedeliveryBackoff;
import org.springframework.beans.factory.DisposableBean;
@@ -46,4 +47,6 @@ public interface PulsarMessageListenerContainer extends SmartLifecycle, Disposab
void setNegativeAckRedeliveryBackoff(RedeliveryBackoff redeliveryBackoff);
void setDeadLetterPolicy(DeadLetterPolicy deadLetterPolicy);
}

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

@@ -30,6 +30,7 @@ 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;
import org.apache.pulsar.client.api.RedeliveryBackoff;
import org.apache.pulsar.client.api.Schema;
@@ -41,28 +42,28 @@ import org.springframework.pulsar.core.PulsarConsumerFactory;
/**
* @author Soby Chacko
* @author Alexander Preuß
*/
public class ConcurrentPulsarMessageListenerContainerTests {
@Test
@SuppressWarnings("unchecked")
void deadLetterPolicyAppliedOnChildContainer() throws Exception {
PulsarListenerMockComponents env = setupListenerMockComponents(SubscriptionType.Shared);
ConcurrentPulsarMessageListenerContainer<String> concurrentContainer = env.concurrentContainer();
DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder().maxRedeliverCount(5).deadLetterTopic("dlq-topic")
.retryLetterTopic("retry-topic").build();
concurrentContainer.setDeadLetterPolicy(deadLetterPolicy);
concurrentContainer.start();
final DefaultPulsarMessageListenerContainer<String> childContainer = concurrentContainer.getContainers().get(0);
assertThat(childContainer.getDeadLetterPolicy()).isEqualTo(deadLetterPolicy);
}
@Test
void nackRedeliveryBackoffAppliedOnChildContainer() throws Exception {
PulsarConsumerFactory<String> pulsarConsumerFactory = mock(PulsarConsumerFactory.class);
Consumer<String> consumer = mock(Consumer.class);
when(pulsarConsumerFactory.createConsumer(any(Schema.class), any(BatchReceivePolicy.class), any(Map.class)))
.thenReturn(consumer);
when(consumer.batchReceive()).thenReturn(mock(Messages.class));
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
pulsarContainerProperties.setSchema(Schema.STRING);
pulsarContainerProperties.setSubscriptionType(SubscriptionType.Shared);
pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener<?>) (cons, msg) -> {
});
ConcurrentPulsarMessageListenerContainer<String> concurrentContainer = new ConcurrentPulsarMessageListenerContainer<>(
pulsarConsumerFactory, pulsarContainerProperties);
PulsarListenerMockComponents env = setupListenerMockComponents(SubscriptionType.Shared);
ConcurrentPulsarMessageListenerContainer<String> concurrentContainer = env.concurrentContainer();
RedeliveryBackoff redeliveryBackoff = MultiplierRedeliveryBackoff.builder().minDelayMs(1000)
.maxDelayMs(5 * 1000).build();
concurrentContainer.setNegativeAckRedeliveryBackoff(redeliveryBackoff);
@@ -76,26 +77,14 @@ public class ConcurrentPulsarMessageListenerContainerTests {
@Test
@SuppressWarnings("unchecked")
void basicConcurrencyTesting() throws Exception {
PulsarConsumerFactory<String> pulsarConsumerFactory = mock(PulsarConsumerFactory.class);
Consumer<String> consumer = mock(Consumer.class);
PulsarListenerMockComponents env = setupListenerMockComponents(SubscriptionType.Failover);
PulsarConsumerFactory<String> pulsarConsumerFactory = env.consumerFactory();
Consumer<String> consumer = env.consumer();
ConcurrentPulsarMessageListenerContainer<String> concurrentContainer = env.concurrentContainer();
when(pulsarConsumerFactory.createConsumer(any(Schema.class), any(BatchReceivePolicy.class), any(Map.class)))
.thenReturn(consumer);
concurrentContainer.setConcurrency(3);
when(consumer.batchReceive()).thenReturn(mock(Messages.class));
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
pulsarContainerProperties.setSchema(Schema.STRING);
pulsarContainerProperties.setSubscriptionType(SubscriptionType.Failover);
pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener<?>) (cons, msg) -> {
});
ConcurrentPulsarMessageListenerContainer<String> container = new ConcurrentPulsarMessageListenerContainer<>(
pulsarConsumerFactory, pulsarContainerProperties);
container.setConcurrency(3);
container.start();
concurrentContainer.start();
await().atMost(Duration.ofSeconds(10)).untilAsserted(() -> verify(pulsarConsumerFactory, times(3))
.createConsumer(any(Schema.class), any(BatchReceivePolicy.class), any(Map.class)));
@@ -103,8 +92,19 @@ public class ConcurrentPulsarMessageListenerContainerTests {
}
@Test
@SuppressWarnings("unchecked")
void exclusiveSubscriptionMustUseSingleThread() throws Exception {
PulsarListenerMockComponents env = setupListenerMockComponents(SubscriptionType.Exclusive);
ConcurrentPulsarMessageListenerContainer<String> concurrentContainer = env.concurrentContainer();
concurrentContainer.setConcurrency(3);
assertThatThrownBy(concurrentContainer::start).isInstanceOf(IllegalStateException.class)
.hasMessage("concurrency > 1 is not allowed on Exclusive subscription type");
}
@SuppressWarnings("unchecked")
private PulsarListenerMockComponents setupListenerMockComponents(SubscriptionType subscriptionType)
throws Exception {
PulsarConsumerFactory<String> pulsarConsumerFactory = mock(PulsarConsumerFactory.class);
Consumer<String> consumer = mock(Consumer.class);
@@ -115,16 +115,18 @@ public class ConcurrentPulsarMessageListenerContainerTests {
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
pulsarContainerProperties.setSchema(Schema.STRING);
pulsarContainerProperties.setSubscriptionType(subscriptionType);
pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener<?>) (cons, msg) -> {
});
ConcurrentPulsarMessageListenerContainer<String> container = new ConcurrentPulsarMessageListenerContainer<>(
ConcurrentPulsarMessageListenerContainer<String> concurrentContainer = new ConcurrentPulsarMessageListenerContainer<>(
pulsarConsumerFactory, pulsarContainerProperties);
container.setConcurrency(3);
return new PulsarListenerMockComponents(pulsarConsumerFactory, consumer, concurrentContainer);
}
assertThatThrownBy(container::start).isInstanceOf(IllegalStateException.class)
.hasMessage("concurrency > 1 is not allowed on Exclusive subscription type");
private record PulsarListenerMockComponents(PulsarConsumerFactory<String> consumerFactory,
Consumer<String> consumer, ConcurrentPulsarMessageListenerContainer<String> concurrentContainer) {
}
}

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

@@ -33,6 +33,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.DeadLetterPolicy;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.RedeliveryBackoff;
@@ -50,6 +51,7 @@ import org.springframework.pulsar.core.PulsarTemplate;
/**
* @author Soby Chacko
* @author Alexander Preuß
*/
class DefaultPulsarMessageListenerContainerTests extends AbstractContainerBaseTests {
@@ -202,4 +204,60 @@ class DefaultPulsarMessageListenerContainerTests extends AbstractContainerBaseTe
pulsarClient.close();
}
@Test
void deadLetterPolicy() throws Exception {
Map<String, Object> 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(1).deadLetterTopic("dlq-topic")
.build();
config.put("deadLetterPolicy", deadLetterPolicy);
final PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(getPulsarBrokerUrl()).build();
final DefaultPulsarConsumerFactory<Integer> 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<Integer> dlqContainer = new DefaultPulsarMessageListenerContainer<>(
pulsarConsumerFactory, dlqContainerProperties);
dlqContainer.start();
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener<Integer>) (consumer, msg) -> {
latch.countDown();
if (msg.getValue() == 5) {
throw new RuntimeException("fail");
}
});
pulsarContainerProperties.setSchema(Schema.INT32);
pulsarContainerProperties.setSubscriptionType(SubscriptionType.Shared);
DefaultPulsarMessageListenerContainer<Integer> container = new DefaultPulsarMessageListenerContainer<>(
pulsarConsumerFactory, pulsarContainerProperties);
container.start();
Map<String, Object> prodConfig = Collections.singletonMap("topicName", "dpmlct-016");
final DefaultPulsarProducerFactory<Integer> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
pulsarClient, prodConfig);
final PulsarTemplate<Integer> 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();
pulsarClient.close();
}
}

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

@@ -29,6 +29,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.DeadLetterPolicy;
import org.apache.pulsar.client.api.Messages;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.RedeliveryBackoff;
@@ -266,6 +267,47 @@ public class PulsarListenerTests extends AbstractContainerBaseTests {
}
@Nested
@ContextConfiguration(classes = DeadLetterPolicyTest.DeadLetterPolicyConfig.class)
class DeadLetterPolicyTest {
private static CountDownLatch latch = new CountDownLatch(2);
private static CountDownLatch dlqLatch = new CountDownLatch(1);
@Test
void pulsarListenerWithDeadLetterPolicy() throws Exception {
pulsarTemplate.send("dlpt-topic-1", "hello");
assertThat(dlqLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
}
@EnablePulsar
@Configuration
static class DeadLetterPolicyConfig {
@PulsarListener(id = "deadLetterPolicyListener", subscriptionName = "deadLetterPolicySubscription",
topics = "dlpt-topic-1", deadLetterPolicy = "deadLetterPolicy", subscriptionType = "Shared",
properties = { "ackTimeoutMillis=1" })
void listen(String msg) {
latch.countDown();
throw new RuntimeException("fail " + msg);
}
@PulsarListener(id = "dlqListener", subscriptionType = "dlqListenerSubscription", topics = "dlq-topic")
void listenDlq(String msg) {
dlqLatch.countDown();
}
@Bean
DeadLetterPolicy deadLetterPolicy() {
return DeadLetterPolicy.builder().maxRedeliverCount(1).deadLetterTopic("dlq-topic").build();
}
}
}
@Nested
class NegativeConcurrency {