Gh 3340

spring-kafka/src/main/java/org/springframework/kafka/event/ConcurrentContainerStoppedEvent.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2018-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.event;
+
+/**
+ * An event published when a container is stopped.
+ *
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class ConcurrentContainerStoppedEvent extends KafkaEvent {
+
+	private static final long serialVersionUID = 1L;
+
+	/**
+	 * Construct an instance with the provided source and container.
+	 * @param source the container instance that generated the event.
+	 * @param container the container or the parent container if the container is a child.
+	 * @since 2.2.1
+	 */
+	public ConcurrentContainerStoppedEvent(Object source, Object container) {
+		super(source, container);
+	}
+
+	@Override
+	public String toString() {
+		return "ContainerStoppedEvent [source=" + getSource() + "]";
+	}
+
+}

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

@@ -50,6 +50,7 @@
 import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.KafkaAdmin;
+import org.springframework.kafka.event.ConcurrentContainerStoppedEvent;
 import org.springframework.kafka.event.ContainerStoppedEvent;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.TopicPartitionOffset;
@@ -652,7 +653,13 @@ public void stop(Runnable callback) {
 
 	@Override
 	public void stopAbnormally(Runnable callback) {
-		doStop(callback, false);
+		this.lifecycleLock.lock();
+		try {
+			doStop(callback, false);
+		}
+		finally {
+			this.lifecycleLock.unlock();
+		}
 		publishContainerStoppedEvent();
 	}
 
@@ -704,6 +711,13 @@ protected void publishContainerStoppedEvent() {
 		}
 	}
 
+	protected void publishConcurrentContainerStoppedEvent() {
+		ApplicationEventPublisher eventPublisher = getApplicationEventPublisher();
+		if (eventPublisher != null) {
+			eventPublisher.publishEvent(new ConcurrentContainerStoppedEvent(this, parentOrThis()));
+		}
+	}
+
 	/**
 	 * Return this or a parent container if this has a parent.
 	 * @return the parent or this.

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

@@ -202,15 +202,24 @@ public boolean isContainerPaused() {
 
 	@Override
 	public boolean isChildRunning() {
-		if (!isRunning()) {
-			return false;
-		}
-		for (MessageListenerContainer container : this.containers) {
-			if (container.isRunning()) {
+		this.lifecycleLock.lock();
+		try {
+			if(isRunning()) {
+				return true;
+			}
+			if(this.containers.isEmpty() && this.stoppedContainers.get() < this.concurrency) {
 				return true;
 			}
+			for (MessageListenerContainer container : this.containers) {
+				if (container.isRunning()) {
+					return true;
+				}
+			}
+			return false;
+		}
+		finally {
+			this.lifecycleLock.unlock();
 		}
-		return false;
 	}
 
 	@Override
@@ -379,9 +388,10 @@ public void childStopped(MessageListenerContainer child, Reason reason) {
 		if (this.reason == null || reason.equals(Reason.AUTH)) {
 			this.reason = reason;
 		}
+		int stoppedContainersCount = this.stoppedContainers.incrementAndGet();
 		if (Reason.AUTH.equals(this.reason)
 				&& getContainerProperties().isRestartAfterAuthExceptions()
-				&& this.concurrency == this.stoppedContainers.incrementAndGet()) {
+				&& this.concurrency == stoppedContainersCount) {
 
 			this.reason = null;
 			this.stoppedContainers.set(0);
@@ -392,6 +402,8 @@ && getContainerProperties().isRestartAfterAuthExceptions()
 				exec = new SimpleAsyncTaskExecutor(getListenerId() + ".authRestart");
 			}
 			exec.execute(this::start);
+		} else if (this.concurrency == stoppedContainersCount) {
+			publishConcurrentContainerStoppedEvent();
 		}
 	}