GH-3338: Fix ConcurrentContainer lifecycle issues

Fixes: #3338

* Fix ConcurrentContainer lifecycle issues

This commit would fix the following issues.

1) 'isChildRunning' API would return true only after all the containers are actually stopped.
2) ConcurrentContainer `start` would be permitted only after all the containers running status is false.
3) ConcurrentContainer `stop` would be permitted if the container is in running status or if previously `stop` API is not called.
4) Move the logic to verify whether to permit the `stop` call to KafkaMessageListenerContainer and ConcurrentMessageListenerContainer.
5) Add 'stopAbnormally' in a Lock.
6) Set the ConcurrentContainer running status to true after `childStarted`
7) Set the ConcurrentContainer running status to false after `childStopped`
8) Call `childStarted` in ConcurrentContainer from KafkaMessageListenerContainer right before publishing ConsumerStartedEvent.

* Fix ContainerGroupSequencer stopParentAndCheckGroup method

This commit would fix the issue when exactly the ConcurrentContainer has to be stopped. As per the earlier logic, running status would not be set to false if any of the container is stopped. This is not correct and modified the logic to set running status to false even if one of the container is stopped. So, it is sufficient to call directly stop API on parent container that would internally check if all the containers are stopped and would execute the callback accordingly.

As per the review comments, this commit reverts the changes related to the ConcurrentContainer `running` status.
Summary of all changes in this PR.

1) 'isChildRunning' API would return true only after all the containers are actually stopped.
2) Add 'stopAbnormally' in a Lock.
3) Call `childStarted` in ConcurrentContainer from KafkaMessageListenerContainer right before publishing ConsumerStartedEvent.

* Revert changes in ContainerGroupSequencer

* Clear previous childContainers during Start call

This commit would change the time at which the childContainers are cleared. Earlier,
childContainers are cleared during `stop` call. But, after this change childContainers
would be cleared only during the next `start` call.

* Clear ChildContainers after all containers stopped.

This commit would include the following changes.

1) Clear all the containers after all the child containers stopped. Previous commit clears only during the fresh start.
2) Publish `ConcurrentContainerStoppedEvent` when the ConcurrentContainer and all the child child containers are stopped. But, previously `ConcurrentContainerStoppedEvent` is emitted when all the containers are stopped.

**Auto-cherry-pick to `3.2.x` & `3.1.x`**

Author: LokeshAlamuri

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

@@ -71,6 +71,7 @@
  * @author Wang Zhiyang
  * @author Soby Chacko
  * @author Sanghyeok An
+ * @author Lokesh Alamuri
  */
 public abstract class AbstractMessageListenerContainer<K, V>
 		implements GenericMessageListenerContainer<K, V>, BeanNameAware, ApplicationEventPublisherAware,
@@ -660,8 +661,14 @@ public void stop(Runnable callback) {
 
 	@Override
 	public void stopAbnormally(Runnable callback) {
-		doStop(callback, false);
-		publishContainerStoppedEvent();
+		this.lifecycleLock.lock();
+		try {
+			doStop(callback, false);
+			publishContainerStoppedEvent();
+		}
+		finally {
+			this.lifecycleLock.unlock();
+		}
 	}
 
 	protected void doStop(Runnable callback) {

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

@@ -204,14 +204,20 @@ public boolean isContainerPaused() {
 
 	@Override
 	public boolean isChildRunning() {
-		if (!isRunning()) {
-			return false;
-		}
-		for (MessageListenerContainer container : this.containers) {
-			if (container.isRunning()) {
+		this.lifecycleLock.lock();
+		try {
+			for (MessageListenerContainer container : this.containers) {
+				if (container.isRunning()) {
+					return true;
+				}
+			}
+			if (this.startedContainers.get() > 0) {
 				return true;
 			}
 		}
+		finally {
+			this.lifecycleLock.unlock();
+		}
 		return false;
 	}
 
@@ -245,7 +251,7 @@ protected void doStart() {
 						+ topicPartitions.length);
 				this.concurrency = topicPartitions.length;
 			}
-			this.startedContainers.set(0);
+			clearState();
 			setRunning(true);
 
 			for (int i = 0; i < this.concurrency; i++) {
@@ -373,37 +379,59 @@ protected void doStop(final Runnable callback, boolean normal) {
 					}
 				}
 			}
-			this.containers.clear();
 			setStoppedNormally(normal);
+			// All the containers are stopped before calling stop API
+			if (this.startedContainers.get() == 0) {
+				publishConcurrentContainerStoppedEvent(this.reason);
+			}
 		}
 	}
 
 	@Override
 	public void childStarted(MessageListenerContainer child) {
-		this.startedContainers.incrementAndGet();
+		this.lifecycleLock.lock();
+		try {
+			if (this.containers.contains(child)) {
+				this.startedContainers.incrementAndGet();
+			}
+		}
+		finally {
+			this.lifecycleLock.unlock();
+		}
 	}
 
 	@Override
 	public void childStopped(MessageListenerContainer child, Reason reason) {
-		if (this.reason == null || reason.equals(Reason.AUTH)) {
-			this.reason = reason;
-		}
-		int startedContainersCount = this.startedContainers.decrementAndGet();
-		if (startedContainersCount == 0) {
-			publishConcurrentContainerStoppedEvent(this.reason);
-			boolean restartContainer = Reason.AUTH.equals(this.reason)
-					&& getContainerProperties().isRestartAfterAuthExceptions();
-			this.reason = null;
-
-			if (restartContainer) {
-				// This has to run on another thread to avoid a deadlock on lifecycleMonitor
-				AsyncTaskExecutor exec = getContainerProperties().getListenerTaskExecutor();
-				if (exec == null) {
-					exec = new SimpleAsyncTaskExecutor(getListenerId() + ".authRestart");
+		this.lifecycleLock.lock();
+		try {
+			if (!this.containers.contains(child)) {
+				return;
+			}
+			if (this.reason == null || reason.equals(Reason.AUTH)) {
+				this.reason = reason;
+			}
+			int startedContainersCount = this.startedContainers.decrementAndGet();
+			if (startedContainersCount == 0) {
+				if (!isRunning()) {
+					this.containers.clear();
+					publishConcurrentContainerStoppedEvent(this.reason);
+				}
+				boolean restartContainer = Reason.AUTH.equals(this.reason)
+						&& getContainerProperties().isRestartAfterAuthExceptions();
+				this.reason = null;
+				if (restartContainer) {
+					// This has to run on another thread to avoid a deadlock on lifecycleMonitor
+					AsyncTaskExecutor exec = getContainerProperties().getListenerTaskExecutor();
+					if (exec == null) {
+						exec = new SimpleAsyncTaskExecutor(getListenerId() + ".authRestart");
+					}
+					exec.execute(this::start);
 				}
-				exec.execute(this::start);
 			}
 		}
+		finally {
+			this.lifecycleLock.unlock();
+		}
 	}
 
 	private void publishConcurrentContainerStoppedEvent(Reason reason) {
@@ -517,6 +545,12 @@ private boolean containsPartition(TopicPartition topicPartition, KafkaMessageLis
 		return assignedPartitions != null && assignedPartitions.contains(topicPartition);
 	}
 
+	private void clearState() {
+		this.containers.clear();
+		this.startedContainers.set(0);
+		this.reason = null;
+	}
+
 	@Override
 	public String toString() {
 		return "ConcurrentMessageListenerContainer [concurrency=" + this.concurrency + ", beanName="

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

@@ -166,6 +166,7 @@
  * @author Christian Mergenthaler
  * @author Mikael Carlstedt
  * @author Borahm Lee
+ * @author Lokesh Alamuri
  */
 public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 		extends AbstractMessageListenerContainer<K, V> implements ConsumerPauseResumeEventPublisher {
@@ -378,7 +379,6 @@ protected void doStart() {
 			}
 		}
 		this.listenerConsumer = new ListenerConsumer(listener, listenerType, observationRegistry);
-		this.thisOrParentContainer.childStarted(this);
 		setRunning(true);
 		this.startLatch = new CountDownLatch(1);
 		this.listenerConsumerFuture = consumerExecutor.submitCompletable(this.listenerConsumer);
@@ -1369,6 +1369,7 @@ protected void initialize() {
 			this.count = 0;
 			this.last = System.currentTimeMillis();
 			initAssignedPartitions();
+			KafkaMessageListenerContainer.this.thisOrParentContainer.childStarted(KafkaMessageListenerContainer.this);
 			publishConsumerStartedEvent();
 		}