GH-53: Fix @KafkaListener autoStartup

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<ContextRefreshedEvent> {
 
 	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();
 		}
 	}

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<ConcurrentMessageListenerContainer<Integer, String>>
-		kafkaManualAckListenerContainerFactory() {
+				kafkaManualAckListenerContainerFactory() {
 			SimpleKafkaListenerContainerFactory<Integer, String> factory = new SimpleKafkaListenerContainerFactory<>();
 			factory.setConsumerFactory(manualConsumerFactory());
 			factory.setAckMode(AckMode.MANUAL_IMMEDIATE);
 			return factory;
 		}
 
+		@Bean
+		public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
+				kafkaAutoStartFalseListenerContainerFactory() {
+			SimpleKafkaListenerContainerFactory<Integer, String> factory = new SimpleKafkaListenerContainerFactory<>();
+			factory.setConsumerFactory(consumerFactory());
+			factory.setAutoStartup(false);
+			return factory;
+		}
+
 		@Bean
 		public ConsumerFactory<Integer, String> 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();