From 0b495cb5ba2d07314017cb4f4101d5b2adacc8f6 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 1 Apr 2016 16:19:53 -0400 Subject: [PATCH] GH-53: Fix @KafkaListener autoStartup --- .../config/KafkaListenerEndpointRegistry.java | 23 +++++++++++----- .../EnableKafkaIntegrationTests.java | 27 ++++++++++++++++++- 2 files changed, 43 insertions(+), 7 deletions(-) diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java b/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java index 24e1e262..e49fc3c1 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java @@ -34,8 +34,10 @@ import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; +import org.springframework.context.ApplicationListener; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.SmartLifecycle; +import org.springframework.context.event.ContextRefreshedEvent; import org.springframework.kafka.listener.MessageListenerContainer; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -62,7 +64,8 @@ import org.springframework.util.StringUtils; * @see MessageListenerContainer * @see KafkaListenerContainerFactory */ -public class KafkaListenerEndpointRegistry implements DisposableBean, SmartLifecycle, ApplicationContextAware { +public class KafkaListenerEndpointRegistry implements DisposableBean, SmartLifecycle, ApplicationContextAware, + ApplicationListener { protected final Log logger = LogFactory.getLog(getClass()); //NOSONAR @@ -73,6 +76,8 @@ public class KafkaListenerEndpointRegistry implements DisposableBean, SmartLifec private ConfigurableApplicationContext applicationContext; + private boolean contextRefreshed; + @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { @@ -228,9 +233,7 @@ public class KafkaListenerEndpointRegistry implements DisposableBean, SmartLifec @Override public void start() { for (MessageListenerContainer listenerContainer : getListenerContainers()) { - if (listenerContainer.isAutoStartup()) { - startIfNecessary(listenerContainer); - } + startIfNecessary(listenerContainer); } } @@ -260,14 +263,22 @@ public class KafkaListenerEndpointRegistry implements DisposableBean, SmartLifec return false; } + + @Override + public void onApplicationEvent(ContextRefreshedEvent event) { + if (event.getApplicationContext().equals(this.applicationContext)) { + this.contextRefreshed = true; + } + } + /** * Start the specified {@link MessageListenerContainer} if it should be started * on startup. * @param listenerContainer the listener container to start. * @see MessageListenerContainer#isAutoStartup() */ - private static void startIfNecessary(MessageListenerContainer listenerContainer) { - if (listenerContainer.isAutoStartup()) { + private void startIfNecessary(MessageListenerContainer listenerContainer) { + if (this.contextRefreshed || listenerContainer.isAutoStartup()) { listenerContainer.start(); } } 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 641c9ad2..b2b1c13d 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 @@ -42,6 +42,7 @@ 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.MessageListenerContainer; import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.converter.StringJsonMessageConverter; @@ -121,6 +122,16 @@ public class EnableKafkaIntegrationTests { assertThat(this.listener.latch5.await(20, TimeUnit.SECONDS)).isTrue(); } + @Test + public void testAutoStartup() throws Exception { + MessageListenerContainer listenerContainer = registry.getListenerContainer("manualStart"); + assertThat(listenerContainer).isNotNull(); + assertThat(listenerContainer.isRunning()).isFalse(); + this.registry.start(); + assertThat(listenerContainer.isRunning()).isTrue(); + listenerContainer.stop(); + } + @Test public void testInterface() throws Exception { template.send("annotated7", 0, "foo"); @@ -184,13 +195,22 @@ public class EnableKafkaIntegrationTests { @Bean public KafkaListenerContainerFactory> - kafkaManualAckListenerContainerFactory() { + kafkaManualAckListenerContainerFactory() { SimpleKafkaListenerContainerFactory factory = new SimpleKafkaListenerContainerFactory<>(); factory.setConsumerFactory(manualConsumerFactory()); factory.setAckMode(AckMode.MANUAL_IMMEDIATE); return factory; } + @Bean + public KafkaListenerContainerFactory> + kafkaAutoStartFalseListenerContainerFactory() { + SimpleKafkaListenerContainerFactory factory = new SimpleKafkaListenerContainerFactory<>(); + factory.setConsumerFactory(consumerFactory()); + factory.setAutoStartup(false); + return factory; + } + @Bean public ConsumerFactory consumerFactory() { return new DefaultKafkaConsumerFactory<>(consumerConfigs()); @@ -275,6 +295,11 @@ public class EnableKafkaIntegrationTests { private Foo foo; + @KafkaListener(id = "manualStart", topics = "manualStart", + containerFactory = "kafkaAutoStartFalseListenerContainerFactory") + public void manualStart(String foo) { + } + @KafkaListener(id = "foo", topics = "annotated1") public void listen1(String foo) { this.latch1.countDown();