Fix ConcurrentMessageListenerContainer.isInExpectedState consistenc…

…y 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`

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();

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();