From 5e5461079ab8b8a27131e9b1a7e9a392dbfe3fd6 Mon Sep 17 00:00:00 2001 From: SamarthAgrawal-002 Date: Mon, 26 Oct 2026 17:25:17 +0530 Subject: [PATCH 1/2] Fix AB-BA deadlock between DockerClientFactory and RyukResourceReaper Under JUnit 5 parallel test execution, threads initializing GenericContainer and ComposeContainer could deadlock due to an inverted lock acquisition order: - Worker 1: DockerClientFactory.$LOCK -> RyukResourceReaper.this - Worker 2: RyukResourceReaper.this -> DockerClientFactory.$LOCK By resolving DockerClientFactory.instance().client() outside the synchronized block in RyukResourceReaper.maybeStart(), we enforce a strict global lock hierarchy where DockerClientFactory is always initialized prior to acquiring the Ryuk monitor, eliminating the cyclic dependency. Fixes #9120. Signed-off-by: SamarthAgrawal-002 --- .../utility/RyukResourceReaper.java | 116 ++++++++++-------- .../RyukResourceReaperDeadlockTest.java | 85 +++++++++++++ 2 files changed, 148 insertions(+), 53 deletions(-) create mode 100644 core/src/test/java/org/testcontainers/RyukResourceReaperDeadlockTest.java diff --git a/core/src/main/java/org/testcontainers/utility/RyukResourceReaper.java b/core/src/main/java/org/testcontainers/utility/RyukResourceReaper.java index 999b1cbbffb..63107cfda24 100644 --- a/core/src/main/java/org/testcontainers/utility/RyukResourceReaper.java +++ b/core/src/main/java/org/testcontainers/utility/RyukResourceReaper.java @@ -70,66 +70,76 @@ public CreateContainerCmd register(GenericContainer container, CreateContaine } @SneakyThrows(InterruptedException.class) - private synchronized void maybeStart() { - if (!started.compareAndSet(false, true)) { + private void maybeStart() { + if (started.get()) { return; } - ryukContainer.start(); - - CountDownLatch ryukScheduledLatch = new CountDownLatch(1); - - String host = ryukContainer.getHost(); - Integer ryukPort = ryukContainer.getFirstMappedPort(); - Thread kiraThread = new Thread( - DockerClientFactory.TESTCONTAINERS_THREAD_GROUP, - () -> { - while (true) { - RYUK_ACK_RATE_LIMITER.doWhenReady(() -> { - int index = 0; - // not set the read timeout, as Ryuk would not send anything unless a new filter is submitted, meaning that we would get a timeout exception pretty quick - try (Socket clientSocket = new Socket()) { - clientSocket.connect(new InetSocketAddress(host, ryukPort), 5 * 1000); - ResourceReaper.FilterRegistry registry = new ResourceReaper.FilterRegistry( - clientSocket.getInputStream(), - clientSocket.getOutputStream() - ); - - synchronized (ResourceReaper.DEATH_NOTE) { - while (true) { - if (ResourceReaper.DEATH_NOTE.size() <= index) { - try { - ResourceReaper.DEATH_NOTE.wait(1_000); - continue; - } catch (InterruptedException e) { - throw new RuntimeException(e); + DockerClientFactory.instance().client(); + + synchronized (this) { + if (!started.compareAndSet(false, true)) { + return; + } + + ryukContainer.start(); + + CountDownLatch ryukScheduledLatch = new CountDownLatch(1); + + String host = ryukContainer.getHost(); + Integer ryukPort = ryukContainer.getFirstMappedPort(); + Thread kiraThread = new Thread( + DockerClientFactory.TESTCONTAINERS_THREAD_GROUP, + () -> { + while (true) { + RYUK_ACK_RATE_LIMITER.doWhenReady(() -> { + int index = 0; + // not set the read timeout, as Ryuk would not send anything unless a new filter is submitted, meaning that we would get a timeout exception pretty quick + try (Socket clientSocket = new Socket()) { + clientSocket.connect(new InetSocketAddress(host, ryukPort), 5 * 1000); + ResourceReaper.FilterRegistry registry = new ResourceReaper.FilterRegistry( + clientSocket.getInputStream(), + clientSocket.getOutputStream() + ); + + synchronized (ResourceReaper.DEATH_NOTE) { + while (true) { + if (ResourceReaper.DEATH_NOTE.size() <= index) { + try { + ResourceReaper.DEATH_NOTE.wait(1_000); + continue; + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + List> filters = ResourceReaper.DEATH_NOTE.get(index); + boolean isAcknowledged = registry.register(filters); + if (isAcknowledged) { + log.debug("Received 'ACK' from Ryuk"); + ryukScheduledLatch.countDown(); + index++; + } else { + log.debug("Didn't receive 'ACK' from Ryuk. Will retry to send filters."); } - } - List> filters = ResourceReaper.DEATH_NOTE.get(index); - boolean isAcknowledged = registry.register(filters); - if (isAcknowledged) { - log.debug("Received 'ACK' from Ryuk"); - ryukScheduledLatch.countDown(); - index++; - } else { - log.debug("Didn't receive 'ACK' from Ryuk. Will retry to send filters."); } } + } catch (IOException e) { + log.warn("Can not connect to Ryuk at {}:{}", host, ryukPort, e); } - } catch (IOException e) { - log.warn("Can not connect to Ryuk at {}:{}", host, ryukPort, e); - } - }); - } - }, - "testcontainers-ryuk" - ); - kiraThread.setDaemon(true); - kiraThread.start(); - // We need to wait before we can start any containers to make sure that we delete them - if (!ryukScheduledLatch.await(TestcontainersConfiguration.getInstance().getRyukTimeout(), TimeUnit.SECONDS)) { - log.error("Timed out waiting for Ryuk container to start. Ryuk's logs:\n{}", ryukContainer.getLogs()); - throw new IllegalStateException(String.format("Could not connect to Ryuk at %s:%s", host, ryukPort)); + }); + } + }, + "testcontainers-ryuk" + ); + kiraThread.setDaemon(true); + kiraThread.start(); + // We need to wait before we can start any containers to make sure that we delete them + if ( + !ryukScheduledLatch.await(TestcontainersConfiguration.getInstance().getRyukTimeout(), TimeUnit.SECONDS) + ) { + log.error("Timed out waiting for Ryuk container to start. Ryuk's logs:\n{}", ryukContainer.getLogs()); + throw new IllegalStateException(String.format("Could not connect to Ryuk at %s:%s", host, ryukPort)); + } } } } diff --git a/core/src/test/java/org/testcontainers/RyukResourceReaperDeadlockTest.java b/core/src/test/java/org/testcontainers/RyukResourceReaperDeadlockTest.java new file mode 100644 index 00000000000..dbd00c29c78 --- /dev/null +++ b/core/src/test/java/org/testcontainers/RyukResourceReaperDeadlockTest.java @@ -0,0 +1,85 @@ +package org.testcontainers; + +import org.junit.jupiter.api.Test; +import org.testcontainers.utility.ResourceReaper; + +import java.util.Collections; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class RyukResourceReaperDeadlockTest { + + @Test + void shouldNotHoldReaperLockWhenCallingDockerClientFactory() { + ResourceReaper reaper = ResourceReaper.instance(); + AtomicBoolean heldLock = new AtomicBoolean(true); + + DockerClientFactory dockerClientFactory = DockerClientFactory.instance(); + RuntimeException simulatedFailure = new RuntimeException("Simulated failure to intercept call") { + @Override + public synchronized Throwable fillInStackTrace() { + heldLock.set(Thread.holdsLock(reaper)); + return this; + } + }; + + dockerClientFactory.cachedClientFailure = simulatedFailure; + try { + assertThatThrownBy(() -> reaper.registerLabelsFilterForCleanup(Collections.singletonMap("k", "v"))) + .isSameAs(simulatedFailure); + + assertThat(heldLock.get()) + .as("RyukResourceReaper monitor must NOT be held when calling DockerClientFactory to prevent deadlocks") + .isFalse(); + } finally { + dockerClientFactory.cachedClientFailure = null; + } + } + + @Test + void concurrentCallersShouldNotDeadlockUnderContention() throws InterruptedException { + ResourceReaper reaper = ResourceReaper.instance(); + DockerClientFactory dockerClientFactory = DockerClientFactory.instance(); + RuntimeException simulatedFailure = new RuntimeException("Simulated client error"); + + dockerClientFactory.cachedClientFailure = simulatedFailure; + try { + int threadCount = 8; + ExecutorService executor = Executors.newFixedThreadPool(threadCount); + CountDownLatch startLatch = new CountDownLatch(1); + CountDownLatch doneLatch = new CountDownLatch(threadCount); + + for (int i = 0; i < threadCount; i++) { + final int id = i; + executor.submit(() -> { + try { + startLatch.await(2, TimeUnit.SECONDS); + if (id % 2 == 0) { + reaper.registerLabelsFilterForCleanup(Collections.singletonMap("test", "val" + id)); + } else { + reaper.getLabels(); + } + } catch (Throwable ignored) {} finally { + doneLatch.countDown(); + } + }); + } + + startLatch.countDown(); + boolean finished = doneLatch.await(5, TimeUnit.SECONDS); + executor.shutdownNow(); + + assertThat(finished) + .as("Concurrent calls to RyukResourceReaper must finish within timeout and never deadlock") + .isTrue(); + } finally { + dockerClientFactory.cachedClientFailure = null; + } + } +} From b0afe882df6610744a6b2bf8170b024dd409c7f2 Mon Sep 17 00:00:00 2001 From: SamarthAgrawal-002 Date: Mon, 26 Oct 2026 18:41:58 +0530 Subject: [PATCH 2/2] Use separate volatile readiness flag for maybeStart fast path Update maybeStart so its fast-path check evaluates a separate volatile readiness flag rather than started. The readiness flag is set only after Ryuk startup and readiness acknowledgment complete, ensuring concurrent threads do not return prematurely before Ryuk is operational. Retain started as the start-once guard inside the monitor so startup failures preserve existing behavior. Signed-off-by: SamarthAgrawal-002 --- .../org/testcontainers/utility/RyukResourceReaper.java | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/core/src/main/java/org/testcontainers/utility/RyukResourceReaper.java b/core/src/main/java/org/testcontainers/utility/RyukResourceReaper.java index 63107cfda24..76e85c01132 100644 --- a/core/src/main/java/org/testcontainers/utility/RyukResourceReaper.java +++ b/core/src/main/java/org/testcontainers/utility/RyukResourceReaper.java @@ -33,6 +33,8 @@ class RyukResourceReaper extends ResourceReaper { private final AtomicBoolean started = new AtomicBoolean(false); + private volatile boolean ready = false; + private final RyukContainer ryukContainer = new RyukContainer(); @Override @@ -71,13 +73,16 @@ public CreateContainerCmd register(GenericContainer container, CreateContaine @SneakyThrows(InterruptedException.class) private void maybeStart() { - if (started.get()) { + if (ready) { return; } DockerClientFactory.instance().client(); synchronized (this) { + if (ready) { + return; + } if (!started.compareAndSet(false, true)) { return; } @@ -140,6 +145,8 @@ private void maybeStart() { log.error("Timed out waiting for Ryuk container to start. Ryuk's logs:\n{}", ryukContainer.getLogs()); throw new IllegalStateException(String.format("Could not connect to Ryuk at %s:%s", host, ryukPort)); } + + ready = true; } } }