Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix race condition in the AbstractMessageListenerContainer lifecycle

Browse Source 
The `stop()` and `childStopped()` in the `ConcurrentMessageListenerContainer` use the same `lifecycleLock`,
however the last one is called from the thread of listener consumer in the child application context.

* Fix `AbstractMessageListenerContainer.stop(wait)` logic to release the `lifecycleLock`
before going to the `latch.await()`.
This still blocks the `stop()` call, but allows the other lifecycle conditions to be fulfilled,
even from different threads
This commit is contained in:
Artem Bilan
2024-09-16 15:31:49 -04:00
parent c6cf896fb1
commit 1c160757af
1 changed files with 27 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -75,7 +75,7 @@ import org.springframework.util.StringUtils;
*/
public abstract class AbstractMessageListenerContainer<K, V>
implements GenericMessageListenerContainer<K, V>, BeanNameAware, ApplicationEventPublisherAware,
ApplicationContextAware {
ApplicationContextAware {
/**
* The default {@link org.springframework.context.SmartLifecycle} phase for listener
@@ -143,7 +143,6 @@ public abstract class AbstractMessageListenerContainer<K, V>
@Nullable
private KafkaAdmin kafkaAdmin;
/**
* Construct an instance with the provided factory and properties.
* @param consumerFactory the factory.
@@ -609,27 +608,35 @@ public abstract class AbstractMessageListenerContainer<K, V>
* @since 2.3.8
*/
public final void stop(boolean wait) {
this.lifecycleLock.lock();
try {
if (isRunning()) {
if (wait) {
final CountDownLatch latch = new CountDownLatch(1);
if (isRunning()) {
if (wait) {
final CountDownLatch latch = new CountDownLatch(1);
this.lifecycleLock.lock();
try {
doStop(latch::countDown);
try {
latch.await(this.containerProperties.getShutdownTimeout(), TimeUnit.MILLISECONDS); // NOSONAR
publishContainerStoppedEvent();
}
catch (@SuppressWarnings("unused") InterruptedException e) {
Thread.currentThread().interrupt();
}
}
else {
doStop(this::publishContainerStoppedEvent);
finally {
this.lifecycleLock.unlock();
}
try {
latch.await(this.containerProperties.getShutdownTimeout(), TimeUnit.MILLISECONDS); // NOSONAR
publishContainerStoppedEvent();
}
catch (@SuppressWarnings("unused") InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
finally {
this.lifecycleLock.unlock();
else {
this.lifecycleLock.lock();
try {
doStop(this::publishContainerStoppedEvent);
}
finally {
this.lifecycleLock.unlock();
}
}
}
}
@@ -706,7 +713,7 @@ public abstract class AbstractMessageListenerContainer<K, V>
@Override
public void onPartitionsLost(Collection<TopicPartition> partitions) {
AbstractMessageListenerContainer.this.logger.info(() ->
getGroupId() + ": partitions lost: " + partitions);
getGroupId() + ": partitions lost: " + partitions);
}
};