GH-475 Fix Listener Container SmartLifecycle phase

Fixes #475

Previously, containers were started in phase 0; thy should be
started in a late phase.

__cherry-pick to 2.0.x, 1.3.x__

(cherry picked from commit 39bf448)

# Conflicts:
#	src/reference/asciidoc/kafka.adoc

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java

@@ -38,6 +38,7 @@
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.context.event.ContextRefreshedEvent;
+import org.springframework.kafka.listener.AbstractMessageListenerContainer;
 import org.springframework.kafka.listener.MessageListenerContainer;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
@@ -72,7 +73,7 @@ public class KafkaListenerEndpointRegistry implements DisposableBean, SmartLifec
 	private final Map<String, MessageListenerContainer> listenerContainers =
 			new ConcurrentHashMap<String, MessageListenerContainer>();
 
-	private int phase = Integer.MAX_VALUE;
+	private int phase = AbstractMessageListenerContainer.DEFAULT_PHASE;
 
 	private ConfigurableApplicationContext applicationContext;
 
@@ -191,10 +192,11 @@ protected MessageListenerContainer createListenerContainer(KafkaListenerEndpoint
 		}
 
 		int containerPhase = listenerContainer.getPhase();
-		if (containerPhase < Integer.MAX_VALUE) {  // a custom phase value
-			if (this.phase < Integer.MAX_VALUE && this.phase != containerPhase) {
-				throw new IllegalStateException("Encountered phase mismatch between container factory definitions: " +
-						this.phase + " vs " + containerPhase);
+		if (listenerContainer.isAutoStartup() &&
+				containerPhase != AbstractMessageListenerContainer.DEFAULT_PHASE) {  // a custom phase value
+			if (this.phase != AbstractMessageListenerContainer.DEFAULT_PHASE && this.phase != containerPhase) {
+				throw new IllegalStateException("Encountered phase mismatch between container "
+						+ "factory definitions: " + this.phase + " vs " + containerPhase);
 			}
 			this.phase = listenerContainer.getPhase();
 		}

spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java

@@ -45,6 +45,11 @@
 public abstract class AbstractMessageListenerContainer<K, V>
 		implements MessageListenerContainer, BeanNameAware, ApplicationEventPublisherAware, SmartLifecycle {
 
+	/**
+	 * The default {@link SmartLifecycle} phase for listener containers {@value #DEFAULT_PHASE}.
+	 */
+	public static final int DEFAULT_PHASE = Integer.MAX_VALUE - 100; // late phase
+
 	protected final Log logger = LogFactory.getLog(this.getClass()); // NOSONAR
 
 	/**
@@ -108,7 +113,7 @@ public enum AckMode {
 
 	private boolean autoStartup = true;
 
-	private int phase = 0;
+	private int phase = DEFAULT_PHASE;
 
 	private volatile boolean running = false;
 

src/reference/asciidoc/kafka.adoc

@@ -461,6 +461,14 @@ public interface Acknowledgment {
 
 This gives the listener control over when offsets are committed.
 
+[[container-auto-startup]]
+====== Listener Container Auto Startup
+
+The listener containers implement `SmartLifecycle` and `autoStartup` is `true` by default; the containers are started in a late phase (`Integer.MAX-VALUE - 100`).
+Other components that implement `SmartLifecycle`, that handle data from listeners, should be started in an earlier phase.
+The `- 100` leaves room for later phases to enable components to be auto-started after the containers.
+
+
 [[kafka-listener-annotation]]
 ===== @KafkaListener Annotation
 
@@ -676,6 +684,31 @@ static class MultiListenerBean {
 }
 ----
 
+[[kafkalistener-lifecycle]]
+===== @KafkaListener Lifecycle Management
+
+The listener containers created for `@KafkaListener` annotations are not beans in the application context.
+Instead, they are registered with an infrastructure bean of type `KafkaListenerEndpointRegistry`.
+This bean manages the containers' lifecycles; it will auto-start any containers that have `autoStartup` set to `true`.
+All containers created by all container factories must be in the same `phase` - see <<container-auto-startup>> for more information.
+You can manage the lifecycle programmatically using the registry; starting/stopping the registry will start/stop all the registered containers.
+Or, you can get a reference to an individual container using its `id` attribute; you can set `autoStartup` on the annotation, which will override the default setting configured into the container factory.
+
+[source, java]
+----
+@Autowired
+private KafkaListenerEndpointRegistry registry;
+
+...
+
+@KafkaListener(id = "myContainer", topics = "myTopic", autoStartup = "false")
+public void listen(...) { ... }
+
+...
+
+    registry.getListenerContainer("myContainer").start();
+----
+
 ===== Filtering Messages
 
 In certain scenarios, such as rebalancing, a message may be redelivered that has already been processed.