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`.

Author: Wzy19930507

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

@@ -389,7 +389,7 @@ && getContainerProperties().isRestartAfterAuthExceptions()
 			if (exec == null) {
 				exec = new SimpleAsyncTaskExecutor(getListenerId() + ".authRestart");
 			}
-			exec.execute(() -> start());
+			exec.execute(this::start);
 		}
 	}
 
@@ -477,16 +477,23 @@ public boolean isPartitionPaused(TopicPartition topicPartition) {
 	public boolean isInExpectedState() {
 		this.lifecycleLock.lock();
 		try {
-			return (isRunning() || isStoppedNormally()) && this.containers
-					.stream()
-					.map(container -> container.isInExpectedState())
-					.allMatch(bool -> Boolean.TRUE.equals(bool));
+			return (isRunning() || isStoppedNormally()) && isAllMatchInExpectedState();
 		}
 		finally {
 			this.lifecycleLock.unlock();
 		}
 	}
 
+	// See https://github.com/spring-projects/spring-kafka/pull/3059.
+	private boolean isAllMatchInExpectedState() {
+		for (KafkaMessageListenerContainer<K, V> kafkaMessageListenerContainer : this.containers) {
+			if (!kafkaMessageListenerContainer.isInExpectedState()) {
+				return false;
+			}
+		}
+		return true;
+	}
+
 	private boolean containsPartition(TopicPartition topicPartition, KafkaMessageListenerContainer<K, V> container) {
 		Collection<TopicPartition> assignedPartitions = container.getAssignedPartitions();
 		return assignedPartitions != null && assignedPartitions.contains(topicPartition);

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