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

GH-42: Ignore Bridged Methods

Browse Source 
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
This commit is contained in:
Gary Russell
2016-03-31 13:36:38 -04:00
committed by Artem Bilan
parent 4bcac8163c
commit 4cf24fbf28
3 changed files with 163 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
build.gradle
View File

@@ -152,6 +152,7 @@ project ('spring-kafka') {
testCompile project (":spring-kafka-test")
testCompile "org.assertj:assertj-core:$assertjVersion"
testCompile "org.springframework:spring-tx:$springVersion"
}
}

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

@@ -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<K, V>
*/
static final String DEFAULT_KAFKA_LISTENER_CONTAINER_FACTORY_BEAN_NAME = "kafkaListenerContainerFactory";
private final Set<Class<?>> nonAnnotatedClasses =
Collections.newSetFromMap(new ConcurrentHashMap<Class<?>, Boolean>(64));
private final Log logger = LogFactory.getLog(getClass());
private KafkaListenerEndpointRegistry endpointRegistry;
@@ -211,27 +221,55 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
@Override
public Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException {
Class<?> targetClass = AopUtils.getTargetClass(bean);
Collection<KafkaListener> classLevelListeners = findListenerAnnotations(targetClass);
final boolean hasClassLevelListeners = classLevelListeners.size() > 0;
final List<Method> multiMethods = new ArrayList<Method>();
ReflectionUtils.doWithMethods(targetClass, new ReflectionUtils.MethodCallback() {
if (!this.nonAnnotatedClasses.contains(bean.getClass())) {
Class<?> targetClass = AopUtils.getTargetClass(bean);
Collection<KafkaListener> classLevelListeners = findListenerAnnotations(targetClass);
final boolean hasClassLevelListeners = classLevelListeners.size() > 0;
final List<Method> multiMethods = new ArrayList<Method>();
Map<Method, Set<KafkaListener>> annotatedMethods = MethodIntrospector.selectMethods(targetClass,
new MethodIntrospector.MetadataLookup<Set<KafkaListener>>() {
@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<KafkaListener> inspect(Method method) {
Set<KafkaListener> listenerMethods = findListenerAnnotations(method);
return (!listenerMethods.isEmpty() ? listenerMethods : null);
}
});
if (hasClassLevelListeners) {
Set<Method> 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<Method, Set<KafkaListener>> 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<K, V>
/*
* AnnotationUtils.getRepeatableAnnotations does not look at interfaces
*/
private Collection<KafkaListener> findListenerAnnotations(Method method) {
private Set<KafkaListener> findListenerAnnotations(Method method) {
Set<KafkaListener> listeners = new HashSet<KafkaListener>();
KafkaListener ann = AnnotationUtils.findAnnotation(method, KafkaListener.class);
if (ann != null) {

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

@@ -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<Integer, String> 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<Integer, String> kmlc =
((ConcurrentMessageListenerContainer<Integer, String>) 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<ConcurrentMessageListenerContainer<Integer, String>>
kafkaListenerContainerFactory() {
@@ -153,13 +172,15 @@ public class EnableKafkaIntegrationTests {
@Bean
public ConsumerFactory<Integer, String> manualConsumerFactory() {
Map<String, Object> configs = consumerConfigs();
configs.put("enable.auto.commit", "false");
configs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
return new DefaultKafkaConsumerFactory<>(configs);
}
@Bean
public Map<String, Object> consumerConfigs() {
return KafkaTestUtils.consumerProps("testAnnot", "true", embeddedKafka);
Map<String, Object> 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<String> ifaceListener() {
return new IfaceListenerImpl();
}
@Bean
public MultiListenerBean multiListener() {
return new MultiListenerBean();
}
@Bean
public ProducerFactory<Integer, String> 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<T> {
void listen(T foo);
}
static class IfaceListenerImpl implements IfaceListener<String> {
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) {
}
}
}