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 a2255b8381
commit b279945084
2 changed files with 31 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -352,7 +352,8 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
Method defaultMethod = null;
for (Method method : multiMethods) {
Method checked = checkProxy(method, bean);
if (AnnotationUtils.findAnnotation(method, KafkaHandler.class).isDefault()) { // NOSONAR never null
KafkaHandler annotation = AnnotationUtils.findAnnotation(method, KafkaHandler.class);
if (annotation != null && annotation.isDefault()) {
final Method toAssert = defaultMethod;
Assert.state(toAssert == null, () -> "Only one @KafkaHandler can be marked 'isDefault', found: "
+ toAssert.toString() + " and " + method.toString());
@@ -363,7 +364,6 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
for (KafkaListener classLevelListener : classLevelListeners) {
MultiMethodKafkaListenerEndpoint<K, V> endpoint =
new MultiMethodKafkaListenerEndpoint<>(checkedMethods, defaultMethod, bean);
endpoint.setBeanFactory(this.beanFactory);
processListener(endpoint, classLevelListener, bean, bean.getClass(), beanName);
}
}
@@ -372,11 +372,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);
}
@@ -459,6 +454,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);
if (StringUtils.hasText(beanRef)) {
this.listenerScope.removeListener(beanRef);

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

@@ -341,14 +341,17 @@ public class EnableKafkaIntegrationTests {
ConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
Consumer<Integer, String> consumer = cf.createConsumer();
embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "annotated8reply");
template.send("annotated8", 0, 1, "foo");
template.send("annotated8", 0, 1, null);
template.flush();
this.template.send("annotated8", 0, 1, "foo");
this.template.send("annotated8", 0, 1, null);
this.template.flush();
assertThat(this.multiListener.latch1.await(60, TimeUnit.SECONDS)).isTrue();
assertThat(this.multiListener.latch2.await(60, TimeUnit.SECONDS)).isTrue();
ConsumerRecord<Integer, String> reply = KafkaTestUtils.getSingleRecord(consumer, "annotated8reply");
assertThat(reply.value()).isEqualTo("OK");
consumer.close();
template.send("annotated8", 0, 1, "junk");
assertThat(this.multiListener.errorLatch.await(60, TimeUnit.SECONDS)).isTrue();
}
@Test
@@ -1259,6 +1262,15 @@ public class EnableKafkaIntegrationTests {
});
}
@Bean
public KafkaListenerErrorHandler consumeMultiMethodException(MultiListenerBean listener) {
return (m, e) -> {
listener.errorLatch.countDown();
return null;
};
}
}
@Component
@@ -1667,16 +1679,23 @@ public class EnableKafkaIntegrationTests {
}
@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) {
this.latch1.countDown();
if ("junk".equals(bar)) {
throw new RuntimeException("intentional");
}
else {
this.latch1.countDown();
}
}
@KafkaHandler