From f55bbbbfddceb0c9ef8af3654cd9c934c90a0909 Mon Sep 17 00:00:00 2001
From: Wzy19930507 <1208931582@qq.com>
Date: Thu, 22 Feb 2024 20:32:20 +0800
Subject: [PATCH] Fix `ConcurrentMessageListenerContainer.isInExpectedState`
 consistency problem

`ConcurrentMessageListenerContainer#isInExpectedState` throw `ConcurrentModificationException` needs simultaneously invoke `KafkaMessageListenerContainer#setStoppedNormally`, `ConcurrentMessageListenerContainer#isInExpectedState`, modify property in stream.

`TestOOMError#testOOMCMLC` throw `ConcurrentModificationException`, when assertThat container.isInExpectedState(), because of container maybe is not in expected state. concurrent container publishes one time, child container publishes concurrency time, `CountDownLatch` needs to modify to 2.

* Fix `ConcurrentMessageListenerContainer.isInExpectedState` consistency problem
* Fix `TestOOMError#testOOMCMLC` throw `ConcurrentModificationException`
---
 .../ConcurrentMessageListenerContainer.java         | 13 +++++++++----
 .../kafka/listener/TestOOMError.java                |  5 +++--
 2 files changed, 12 insertions(+), 6 deletions(-)

diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java
index 45d9dd85a6..eb350e96d8 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java
@@ -368,7 +368,13 @@ protected void doStop(final Runnable callback, boolean normal) {
 				}
 			}
 			this.containers.clear();
-			setStoppedNormally(normal);
+			this.lifecycleLock.lock();
+			try {
+				setStoppedNormally(normal);
+			}
+			finally {
+				this.lifecycleLock.unlock();
+			}
 		}
 	}
 
@@ -389,7 +395,7 @@ && getContainerProperties().isRestartAfterAuthExceptions()
 			if (exec == null) {
 				exec = new SimpleAsyncTaskExecutor(getListenerId() + ".authRestart");
 			}
-			exec.execute(() -> start());
+			exec.execute(this::start);
 		}
 	}
 
@@ -479,8 +485,7 @@ public boolean isInExpectedState() {
 		try {
 			return (isRunning() || isStoppedNormally()) && this.containers
 					.stream()
-					.map(container -> container.isInExpectedState())
-					.allMatch(bool -> Boolean.TRUE.equals(bool));
+					.allMatch(KafkaMessageListenerContainer::isInExpectedState);
 		}
 		finally {
 			this.lifecycleLock.unlock();
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/TestOOMError.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/TestOOMError.java
index 0b8c7627db..29f0bafc94 100644
--- a/spring-kafka/src/test/java/org/springframework/kafka/listener/TestOOMError.java
+++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/TestOOMError.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2022 the original author or authors.
+ * Copyright 2018-2024 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.
@@ -77,7 +77,8 @@ public void testOOMCMLC() throws Exception {
 		containerProps.setClientId("clientId");
 		ConcurrentMessageListenerContainer<Integer, String> container =
 				new ConcurrentMessageListenerContainer<>(cf, containerProps);
-		CountDownLatch stopLatch = new CountDownLatch(1);
+		// concurrent container publishes one time, child container publishes concurrency time.
+		CountDownLatch stopLatch = new CountDownLatch(2);
 		container.setApplicationEventPublisher(e -> {
 			if (e instanceof ContainerStoppedEvent) {
 				stopLatch.countDown();