From 4cf24fbf28a0b8da390dba20c37a361df81cb0b4 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 31 Mar 2016 13:36:38 -0400 Subject: [PATCH] GH-42: Ignore Bridged Methods Resolves #42 Convert to MethodIntrospector Polishing * Use more simple `selectMethods` for `boolean` case * Add `spring-tx` dependency * add `@Transactional` test-case to be sure that our job around proxies is fine --- build.gradle | 1 + ...kaListenerAnnotationBeanPostProcessor.java | 76 +++++++--- .../EnableKafkaIntegrationTests.java | 130 ++++++++++++++---- 3 files changed, 163 insertions(+), 44 deletions(-) diff --git a/build.gradle b/build.gradle index 8300f742..3ef04359 100644 --- a/build.gradle +++ b/build.gradle @@ -152,6 +152,7 @@ project ('spring-kafka') { testCompile project (":spring-kafka-test") testCompile "org.assertj:assertj-core:$assertjVersion" + testCompile "org.springframework:spring-tx:$springVersion" } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java b/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java index 48acb379..2e95f5b4 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java @@ -20,13 +20,18 @@ import java.lang.reflect.Method; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; import java.util.regex.Pattern; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + import org.springframework.aop.framework.Advised; import org.springframework.aop.support.AopUtils; import org.springframework.beans.BeansException; @@ -42,6 +47,7 @@ import org.springframework.beans.factory.config.BeanPostProcessor; import org.springframework.beans.factory.config.ConfigurableBeanFactory; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.context.expression.StandardBeanExpressionResolver; +import org.springframework.core.MethodIntrospector; import org.springframework.core.Ordered; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.kafka.config.KafkaListenerConfigUtils; @@ -96,6 +102,10 @@ public class KafkaListenerAnnotationBeanPostProcessor */ static final String DEFAULT_KAFKA_LISTENER_CONTAINER_FACTORY_BEAN_NAME = "kafkaListenerContainerFactory"; + private final Set> nonAnnotatedClasses = + Collections.newSetFromMap(new ConcurrentHashMap, Boolean>(64)); + + private final Log logger = LogFactory.getLog(getClass()); private KafkaListenerEndpointRegistry endpointRegistry; @@ -211,27 +221,55 @@ public class KafkaListenerAnnotationBeanPostProcessor @Override public Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException { - Class targetClass = AopUtils.getTargetClass(bean); - Collection classLevelListeners = findListenerAnnotations(targetClass); - final boolean hasClassLevelListeners = classLevelListeners.size() > 0; - final List multiMethods = new ArrayList(); - ReflectionUtils.doWithMethods(targetClass, new ReflectionUtils.MethodCallback() { + if (!this.nonAnnotatedClasses.contains(bean.getClass())) { + Class targetClass = AopUtils.getTargetClass(bean); + Collection classLevelListeners = findListenerAnnotations(targetClass); + final boolean hasClassLevelListeners = classLevelListeners.size() > 0; + final List multiMethods = new ArrayList(); + Map> annotatedMethods = MethodIntrospector.selectMethods(targetClass, + new MethodIntrospector.MetadataLookup>() { - @Override - public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { - for (KafkaListener kafkaListener : findListenerAnnotations(method)) { - processKafkaListener(kafkaListener, method, bean, beanName); - } - if (hasClassLevelListeners) { - KafkaHandler kafkaHandler = AnnotationUtils.findAnnotation(method, KafkaHandler.class); - if (kafkaHandler != null) { - multiMethods.add(method); - } + @Override + public Set inspect(Method method) { + Set listenerMethods = findListenerAnnotations(method); + return (!listenerMethods.isEmpty() ? listenerMethods : null); + } + + }); + if (hasClassLevelListeners) { + Set methodsWithHandler = MethodIntrospector.selectMethods(targetClass, + new ReflectionUtils.MethodFilter() { + + @Override + public boolean matches(Method method) { + return AnnotationUtils.findAnnotation(method, KafkaHandler.class) != null; + } + + }); + multiMethods.addAll(methodsWithHandler); + } + if (annotatedMethods.isEmpty()) { + this.nonAnnotatedClasses.add(bean.getClass()); + if (this.logger.isTraceEnabled()) { + this.logger.trace("No @KafkaListener annotations found on bean type: " + bean.getClass()); } } - }); - if (hasClassLevelListeners) { - processMultiMethodListeners(classLevelListeners, multiMethods, bean, beanName); + else { + // Non-empty set of methods + for (Map.Entry> entry : annotatedMethods.entrySet()) { + Method method = entry.getKey(); + for (KafkaListener listener : entry.getValue()) { + processKafkaListener(listener, method, bean, beanName); + } + } + if (this.logger.isDebugEnabled()) { + this.logger.debug(annotatedMethods.size() + " @KafkaListener methods processed on bean '" + + beanName + "': " + annotatedMethods); + } + } + if (hasClassLevelListeners) { + processMultiMethodListeners(classLevelListeners, multiMethods, bean, beanName); + } } return bean; } @@ -255,7 +293,7 @@ public class KafkaListenerAnnotationBeanPostProcessor /* * AnnotationUtils.getRepeatableAnnotations does not look at interfaces */ - private Collection findListenerAnnotations(Method method) { + private Set findListenerAnnotations(Method method) { Set listeners = new HashSet(); KafkaListener ann = AnnotationUtils.findAnnotation(method, KafkaListener.class); if (ann != null) { diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java index ab206a1c..3f6acc6f 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java @@ -22,10 +22,12 @@ import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.junit.ClassRule; import org.junit.Test; import org.junit.runner.RunWith; +import org.mockito.Mockito; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; @@ -40,8 +42,6 @@ import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode; import org.springframework.kafka.listener.ConcurrentMessageListenerContainer; -import org.springframework.kafka.listener.KafkaMessageListenerContainer; -import org.springframework.kafka.listener.MessageListenerContainer; import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.test.rule.KafkaEmbedded; @@ -51,6 +51,9 @@ import org.springframework.messaging.handler.annotation.Payload; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.annotation.EnableTransactionManagement; +import org.springframework.transaction.annotation.Transactional; /** * @author Gary Russell @@ -62,13 +65,19 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @DirtiesContext public class EnableKafkaIntegrationTests { - @ClassRule - public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, "annotated1", "annotated2", "annotated3", - "annotated4", "annotated5", "annotated6"); - @Autowired public Listener listener; + @ClassRule + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, "annotated1", "annotated2", "annotated3", + "annotated4", "annotated5", "annotated6", "annotated7", "annotated8", "annotated9"); + + @Autowired + public IfaceListenerImpl ifaceListener; + + @Autowired + public MultiListenerBean multiListener; + @Autowired public KafkaTemplate template; @@ -77,12 +86,10 @@ public class EnableKafkaIntegrationTests { @Test public void testSimple() throws Exception { - waitListening("foo"); template.send("annotated1", 0, "foo"); template.flush(); assertThat(this.listener.latch1.await(10, TimeUnit.SECONDS)).isTrue(); - waitListening("bar"); template.send("annotated2", 0, 123, "foo"); template.flush(); assertThat(this.listener.latch2.await(10, TimeUnit.SECONDS)).isTrue(); @@ -90,44 +97,56 @@ public class EnableKafkaIntegrationTests { assertThat(this.listener.partition).isNotNull(); assertThat(this.listener.topic).isEqualTo("annotated2"); - waitListening("baz"); template.send("annotated3", 0, "foo"); template.flush(); assertThat(this.listener.latch3.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(this.listener.record.value()).isEqualTo("foo"); - waitListening("qux"); template.send("annotated4", 0, "foo"); template.flush(); assertThat(this.listener.latch4.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(this.listener.record.value()).isEqualTo("foo"); assertThat(this.listener.ack).isNotNull(); - waitListening("fiz"); template.send("annotated5", 0, 0, "foo"); template.send("annotated5", 1, 0, "bar"); template.send("annotated6", 0, 0, "baz"); template.send("annotated6", 1, 0, "qux"); template.flush(); - assertThat(this.listener.latch5.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.listener.latch5.await(20, TimeUnit.SECONDS)).isTrue(); } - private void waitListening(String id) throws InterruptedException { - MessageListenerContainer container = registry.getListenerContainer(id); - @SuppressWarnings("unchecked") - KafkaMessageListenerContainer kmlc = - ((ConcurrentMessageListenerContainer) container).getContainers().get(0); - int n = 0; - while (n++ < 6000 && (kmlc.getAssignedPartitions() == null || kmlc.getAssignedPartitions().size() == 0)) { - Thread.sleep(100); - } - assertThat(kmlc.getAssignedPartitions().size()).isGreaterThan(0); + @Test + public void testInterface() throws Exception { + template.send("annotated7", 0, "foo"); + template.flush(); + assertThat(this.ifaceListener.getLatch1().await(20, TimeUnit.SECONDS)).isTrue(); + } + + @Test + public void testMulti() throws Exception { + template.send("annotated8", 0, "foo"); + template.flush(); + assertThat(this.multiListener.latch1.await(20, TimeUnit.SECONDS)).isTrue(); + } + + @Test + public void testTx() throws Exception { + template.send("annotated9", 0, "foo"); + template.flush(); + assertThat(this.ifaceListener.getLatch2().await(20, TimeUnit.SECONDS)).isTrue(); } @Configuration @EnableKafka + @EnableTransactionManagement(proxyTargetClass = true) public static class Config { + @Bean + public PlatformTransactionManager transactionManager() { + return Mockito.mock(PlatformTransactionManager.class); + } + @Bean public KafkaListenerContainerFactory> kafkaListenerContainerFactory() { @@ -153,13 +172,15 @@ public class EnableKafkaIntegrationTests { @Bean public ConsumerFactory manualConsumerFactory() { Map configs = consumerConfigs(); - configs.put("enable.auto.commit", "false"); + configs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); return new DefaultKafkaConsumerFactory<>(configs); } @Bean public Map consumerConfigs() { - return KafkaTestUtils.consumerProps("testAnnot", "true", embeddedKafka); + Map consumerProps = KafkaTestUtils.consumerProps("testAnnot", "true", embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + return consumerProps; } @Bean @@ -167,6 +188,16 @@ public class EnableKafkaIntegrationTests { return new Listener(); } + @Bean + public IfaceListener ifaceListener() { + return new IfaceListenerImpl(); + } + + @Bean + public MultiListenerBean multiListener() { + return new MultiListenerBean(); + } + @Bean public ProducerFactory producerFactory() { return new DefaultKafkaProducerFactory<>(producerConfigs()); @@ -184,7 +215,7 @@ public class EnableKafkaIntegrationTests { } - public static class Listener { + static class Listener { private final CountDownLatch latch1 = new CountDownLatch(1); @@ -246,4 +277,53 @@ public class EnableKafkaIntegrationTests { } + interface IfaceListener { + + void listen(T foo); + + } + + static class IfaceListenerImpl implements IfaceListener { + + private final CountDownLatch latch1 = new CountDownLatch(1); + + private final CountDownLatch latch2 = new CountDownLatch(1); + + @Override + @KafkaListener(id = "ifc", topics = "annotated7") + public void listen(String foo) { + latch1.countDown(); + } + + @KafkaListener(topics = "annotated9") + @Transactional + public void listenTx(String foo) { + latch2.countDown(); + } + + public CountDownLatch getLatch1() { + return latch1; + } + + public CountDownLatch getLatch2() { + return latch2; + } + } + + @KafkaListener(topics = "annotated8") + static class MultiListenerBean { + + private final CountDownLatch latch1 = new CountDownLatch(1); + + @KafkaHandler + public void bar(String bar) { + latch1.countDown(); + } + + public void foo(String bar) { + } + + } + + }