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