GH-1102: Fix KLEndpointRegistry SmartLifecycle

Resolves #1102

Add `running` boolean for `isRunning()`.

Previously the registry returned `true` if at least one of its
containers had been previously started manually.

Author: garyrussell

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

@@ -79,6 +79,7 @@ public class KafkaListenerEndpointRegistry implements DisposableBean, SmartLifec
 
 	private boolean contextRefreshed;
 
+	private volatile boolean running;
 
 	@Override
 	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
@@ -254,13 +255,15 @@ public void start() {
 		for (MessageListenerContainer listenerContainer : getListenerContainers()) {
 			startIfNecessary(listenerContainer);
 		}
+		this.running = true;
 	}
 
 	@Override
 	public void stop() {
 		for (MessageListenerContainer listenerContainer : getListenerContainers()) {
 			listenerContainer.stop();
 		}
+		this.running = false;
 	}
 
 	@Override
@@ -279,12 +282,7 @@ public void stop(Runnable callback) {
 
 	@Override
 	public boolean isRunning() {
-		for (MessageListenerContainer listenerContainer : getListenerContainers()) {
-			if (listenerContainer.isRunning()) {
-				return true;
-			}
-		}
-		return false;
+		return this.running;
 	}