GH-475 Fix Listener Container SmartLifecycle phase

Fixes https://github.com/spring-projects/spring-kafka/issues/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__

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

@@ -38,6 +38,7 @@ 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.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 @@ public class KafkaListenerEndpointRegistry implements DisposableBean, SmartLifec
 		}
 
 		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

@@ -46,6 +46,11 @@ import org.springframework.util.Assert;
 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
 
 	/**
@@ -109,7 +114,7 @@ public abstract class AbstractMessageListenerContainer<K, V>
 
 	private boolean autoStartup = true;
 
-	private int phase = 0;
+	private int phase = DEFAULT_PHASE;
 
 	private volatile boolean running = false;
 

src/reference/asciidoc/kafka.adoc

@@ -565,6 +565,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
 
@@ -779,6 +787,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();
+----
+
 [[rebalance-listeners]]
 ===== Rebalance Listeners