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

Process errorHandler in class level KafkaListener

Browse Source 
The `errorHandler` attributed has been missed from the
`KafkaListenerAnnotationBeanPostProcessor.processMultiMethodListeners()`
 logic.

* Move `errorHandler` and `BeanFactory` population into the
`processListener()` method in the `KafkaListenerAnnotationBeanPostProcessor`

**Cherry-pick to 2.1.x, 2.0.x & 1.3.x**
This commit is contained in:
Artem Bilan
2018-12-14 13:11:30 -05:00
committed by Gary Russell
parent 87ec5829fd
commit 8fe9406c2f
2 changed files with 24 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java
View File

@@ -325,7 +325,6 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
for (KafkaListener classLevelListener : classLevelListeners) {
MultiMethodKafkaListenerEndpoint<K, V> endpoint = new MultiMethodKafkaListenerEndpoint<>(checkedMethods,
bean);
endpoint.setBeanFactory(this.beanFactory);
processListener(endpoint, classLevelListener, bean, bean.getClass(), beanName);
}
}
@@ -334,11 +333,6 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
Method methodToUse = checkProxy(method, bean);
MethodKafkaListenerEndpoint<K, V> endpoint = new MethodKafkaListenerEndpoint<>();
endpoint.setMethod(methodToUse);
endpoint.setBeanFactory(this.beanFactory);
String errorHandlerBeanName = resolveExpressionAsString(kafkaListener.errorHandler(), "errorHandler");
if (StringUtils.hasText(errorHandlerBeanName)) {
endpoint.setErrorHandler(this.beanFactory.getBean(errorHandlerBeanName, KafkaListenerErrorHandler.class));
}
processListener(endpoint, kafkaListener, bean, methodToUse, beanName);
}
@@ -405,6 +399,11 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
}
}
endpoint.setBeanFactory(this.beanFactory);
String errorHandlerBeanName = resolveExpressionAsString(kafkaListener.errorHandler(), "errorHandler");
if (StringUtils.hasText(errorHandlerBeanName)) {
endpoint.setErrorHandler(this.beanFactory.getBean(errorHandlerBeanName, KafkaListenerErrorHandler.class));
}
this.registrar.registerEndpoint(endpoint, factory);
}

21
spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java
View File

@@ -281,6 +281,8 @@ public class EnableKafkaIntegrationTests {
template.flush();
assertThat(this.multiListener.latch1.await(60, TimeUnit.SECONDS)).isTrue();
assertThat(this.multiListener.latch2.await(60, TimeUnit.SECONDS)).isTrue();
template.send("annotated8", 0, 1, "junk");
assertThat(this.multiListener.errorLatch.await(60, TimeUnit.SECONDS)).isTrue();
}
@Test
@@ -1273,18 +1275,33 @@ public class EnableKafkaIntegrationTests {
return latch2;
}
@Bean
public KafkaListenerErrorHandler consumeMultiMethodException(MultiListenerBean listener) {
return (m, e) -> {
listener.errorLatch.countDown();
return null;
};
}
}
@KafkaListener(id = "multi", topics = "annotated8")
@KafkaListener(id = "multi", topics = "annotated8", errorHandler = "consumeMultiMethodException")
static class MultiListenerBean {
private final CountDownLatch latch1 = new CountDownLatch(1);
private final CountDownLatch latch2 = new CountDownLatch(1);
private final CountDownLatch errorLatch = new CountDownLatch(1);
@KafkaHandler
public void bar(@NonNull String bar) {
latch1.countDown();
if ("junk".equals(bar)) {
throw new RuntimeException("intentional");
}
else {
this.latch1.countDown();
}
}
@KafkaHandler