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

More Pulsar Consumer Error handling Changes (#107)

Browse Source 
This PR introduces the ability for PulsarListener annotation
to provide the name of the PulsarConsumerErrorHandler bean.
When such a bean name is provided, the framework properly scans
for the bean and delegates it to downstream message listener container.
This commit is contained in:
Soby Chacko
2022-09-12 16:50:26 -04:00
committed by GitHub
parent df00e9b67a
commit 4fd61b9b2d
9 changed files with 109 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java
View File

@@ -162,7 +162,7 @@ class PulsarAutoConfigurationTests {
@Test
void customPulsarListenerAnnotationBeanPostProcessorIsRespected() {
PulsarListenerAnnotationBeanPostProcessor<String, String> listenerAnnotationBeanPostProcessor = mock(
PulsarListenerAnnotationBeanPostProcessor<String> listenerAnnotationBeanPostProcessor = mock(
PulsarListenerAnnotationBeanPostProcessor.class);
this.contextRunner
.withBean("org.springframework.pulsar.config.internalPulsarListenerAnnotationProcessor",

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

@@ -186,4 +186,12 @@ public @interface PulsarListener {
*/
AckMode ackMode() default AckMode.BATCH;
/**
* The bean name or a 'SpEL' expression that resolves to a
* {@link org.springframework.pulsar.listener.PulsarConsumerErrorHandler} which is
* used as a Spring provided mechanism to handle errors from processing the message.
* @return the bean name for the consumer error handler or an empty string.
*/
String pulsarConsumerErrorHandler() default "";
}

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

@@ -85,6 +85,7 @@ import org.springframework.pulsar.config.PulsarListenerContainerFactory;
import org.springframework.pulsar.config.PulsarListenerEndpoint;
import org.springframework.pulsar.config.PulsarListenerEndpointRegistrar;
import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
import org.springframework.pulsar.listener.PulsarConsumerErrorHandler;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
@@ -107,8 +108,7 @@ import org.springframework.validation.Validator;
* fine-grained control over endpoints registration. See {@link EnablePulsar} Javadoc for
* complete usage details.
*
* @param <K> the key type.
* @param <V> the value type.
* @param <V> the payload type.
* @author Soby Chacko
* @author Chris Bono
* @author Alexander Preuß
@@ -120,7 +120,7 @@ import org.springframework.validation.Validator;
* @see PulsarListenerEndpoint
* @see MethodPulsarListenerEndpoint
*/
public class PulsarListenerAnnotationBeanPostProcessor<K, V>
public class PulsarListenerAnnotationBeanPostProcessor<V>
implements BeanPostProcessor, Ordered, ApplicationContextAware, InitializingBean, SmartInitializingSingleton {
private final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass()));
@@ -366,6 +366,24 @@ public class PulsarListenerAnnotationBeanPostProcessor<K, V>
resolveNegativeAckRedeliveryBackoff(endpoint, pulsarListener);
resolveDeadLetterPolicy(endpoint, pulsarListener);
resolvePulsarConsumerErrorHandler(endpoint, pulsarListener);
}
@SuppressWarnings({ "rawtypes" })
private void resolvePulsarConsumerErrorHandler(MethodPulsarListenerEndpoint<?> endpoint,
PulsarListener pulsarListener) {
Object pulsarConsumerErrorHandler = resolveExpression(pulsarListener.pulsarConsumerErrorHandler());
if (pulsarConsumerErrorHandler instanceof PulsarConsumerErrorHandler) {
endpoint.setPulsarConsumerErrorHandler((PulsarConsumerErrorHandler) pulsarConsumerErrorHandler);
}
else {
String pulsarConsumerErrorHandlerBeanName = resolveExpressionAsString(
pulsarListener.pulsarConsumerErrorHandler(), "pulsarConsumerErrorHandler");
if (StringUtils.hasText(pulsarConsumerErrorHandlerBeanName)) {
endpoint.setPulsarConsumerErrorHandler(
this.beanFactory.getBean(pulsarConsumerErrorHandlerBeanName, PulsarConsumerErrorHandler.class));
}
}
}
private void resolveNegativeAckRedeliveryBackoff(MethodPulsarListenerEndpoint<?> endpoint,

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

@@ -46,6 +46,7 @@ import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
import org.springframework.pulsar.core.SchemaUtils;
import org.springframework.pulsar.listener.Acknowledgement;
import org.springframework.pulsar.listener.ConcurrentPulsarMessageListenerContainer;
import org.springframework.pulsar.listener.PulsarConsumerErrorHandler;
import org.springframework.pulsar.listener.PulsarContainerProperties;
import org.springframework.pulsar.listener.PulsarMessageListenerContainer;
import org.springframework.pulsar.listener.adapter.HandlerAdapter;
@@ -83,6 +84,9 @@ public class MethodPulsarListenerEndpoint<V> extends AbstractPulsarListenerEndpo
private DeadLetterPolicy deadLetterPolicy;
@SuppressWarnings("rawtypes")
private PulsarConsumerErrorHandler pulsarConsumerErrorHandler;
public void setBean(Object bean) {
this.bean = bean;
}
@@ -189,6 +193,7 @@ public class MethodPulsarListenerEndpoint<V> extends AbstractPulsarListenerEndpo
container.setNegativeAckRedeliveryBackoff(this.negativeAckRedeliveryBackoff);
container.setDeadLetterPolicy(this.deadLetterPolicy);
container.setPulsarConsumerErrorHandler(this.pulsarConsumerErrorHandler);
return messageListener;
}
@@ -271,4 +276,9 @@ public class MethodPulsarListenerEndpoint<V> extends AbstractPulsarListenerEndpo
this.deadLetterPolicy = deadLetterPolicy;
}
@SuppressWarnings("rawtypes")
public void setPulsarConsumerErrorHandler(PulsarConsumerErrorHandler pulsarConsumerErrorHandler) {
this.pulsarConsumerErrorHandler = pulsarConsumerErrorHandler;
}
}

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

@@ -65,7 +65,7 @@ public abstract class AbstractPulsarMessageListenerContainer<T> implements Pulsa
protected DeadLetterPolicy deadLetterPolicy;
private PulsarConsumerErrorHandler<T> pulsarConsumerErrorHandler;
protected PulsarConsumerErrorHandler<T> pulsarConsumerErrorHandler;
@SuppressWarnings("unchecked")
protected AbstractPulsarMessageListenerContainer(PulsarConsumerFactory<? super T> pulsarConsumerFactory,
@@ -204,7 +204,8 @@ public abstract class AbstractPulsarMessageListenerContainer<T> implements Pulsa
return this.pulsarConsumerErrorHandler;
}
public void setPulsarConsumerErrorHandler(PulsarConsumerErrorHandler<T> pulsarConsumerErrorHandler) {
@SuppressWarnings({ "rawtypes", "unchecked" })
public void setPulsarConsumerErrorHandler(PulsarConsumerErrorHandler pulsarConsumerErrorHandler) {
this.pulsarConsumerErrorHandler = pulsarConsumerErrorHandler;
}

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

@@ -111,6 +111,7 @@ public class ConcurrentPulsarMessageListenerContainer<T> extends AbstractPulsarM
}
container.setNegativeAckRedeliveryBackoff(this.negativeAckRedeliveryBackoff);
container.setDeadLetterPolicy(this.deadLetterPolicy);
container.setPulsarConsumerErrorHandler(this.pulsarConsumerErrorHandler);
}
@Override

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

@@ -49,4 +49,7 @@ public interface PulsarMessageListenerContainer extends SmartLifecycle, Disposab
void setDeadLetterPolicy(DeadLetterPolicy deadLetterPolicy);
@SuppressWarnings("rawtypes")
void setPulsarConsumerErrorHandler(PulsarConsumerErrorHandler pulsarConsumerErrorHandler);
}

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

@@ -39,6 +39,7 @@ import org.apache.pulsar.client.impl.MultiplierRedeliveryBackoff;
import org.junit.jupiter.api.Test;
import org.springframework.pulsar.core.PulsarConsumerFactory;
import org.springframework.util.backoff.BackOff;
/**
* @author Soby Chacko
@@ -74,6 +75,22 @@ public class ConcurrentPulsarMessageListenerContainerTests {
assertThat(childContainer.getNegativeAckRedeliveryBackoff()).isEqualTo(redeliveryBackoff);
}
@Test
@SuppressWarnings({ "unchecked", "rawtypes" })
void pulsarConsumerErrorHandlerAppliedOnChildContainer() throws Exception {
PulsarListenerMockComponents env = setupListenerMockComponents(SubscriptionType.Shared);
ConcurrentPulsarMessageListenerContainer<String> concurrentContainer = env.concurrentContainer();
PulsarConsumerErrorHandler<String> pulsarConsumerErrorHandler = new DefaultPulsarConsumerErrorHandler(
mock(PulsarMessageRecovererFactory.class), mock(BackOff.class));
concurrentContainer.setPulsarConsumerErrorHandler(pulsarConsumerErrorHandler);
concurrentContainer.start();
final DefaultPulsarMessageListenerContainer<String> childContainer = concurrentContainer.getContainers().get(0);
assertThat(childContainer.getPulsarConsumerErrorHandler()).isEqualTo(pulsarConsumerErrorHandler);
}
@Test
@SuppressWarnings("unchecked")
void basicConcurrencyTesting() throws Exception {

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

@@ -66,6 +66,7 @@ import org.springframework.pulsar.core.PulsarTopic;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.util.backoff.FixedBackOff;
/**
* @author Soby Chacko
@@ -279,6 +280,50 @@ public class PulsarListenerTests extends AbstractContainerBaseTests {
}
@Nested
@ContextConfiguration(classes = PulsarConsumerErrorHandlerTest.PulsarConsumerErrorHandlerConfig.class)
class PulsarConsumerErrorHandlerTest {
private static CountDownLatch pulsarConsumerErrorHandlerLatch = new CountDownLatch(11);
private static CountDownLatch dltLatch = new CountDownLatch(1);
@Test
void pulsarListenerWithNackRedeliveryBackoff(@Autowired PulsarListenerEndpointRegistry registry)
throws Exception {
pulsarTemplate.send("pceht-topic", "hello john doe");
assertThat(pulsarConsumerErrorHandlerLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(dltLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
@EnablePulsar
@Configuration
static class PulsarConsumerErrorHandlerConfig {
@PulsarListener(id = "pceht-id", subscriptionName = "pceht-subscription", topics = "pceht-topic",
pulsarConsumerErrorHandler = "pulsarConsumerErrorHandler")
void listen(String msg) {
pulsarConsumerErrorHandlerLatch.countDown();
throw new RuntimeException("fail " + msg);
}
@PulsarListener(id = "pceh-dltListener", subscriptionType = "dltListenerSubscription",
topics = "pceht-topic-pceht-subscription-DLT")
void listenDlq(String msg) {
dltLatch.countDown();
}
@Bean
public PulsarConsumerErrorHandler<String> pulsarConsumerErrorHandler(
PulsarTemplate<String> pulsarTemplate) {
return new DefaultPulsarConsumerErrorHandler<>(
new PulsarDeadLetterPublishingRecoverer<>(pulsarTemplate), new FixedBackOff(100, 10));
}
}
}
@Nested
@ContextConfiguration(classes = DeadLetterPolicyTest.DeadLetterPolicyConfig.class)
class DeadLetterPolicyTest {