This is an automated email from the ASF dual-hosted git repository.
chrisdutz pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
The following commit(s) were added to refs/heads/develop by this push:
new be5dc80c91 fix: Fixed a potential race condition in the
connection-cache.
be5dc80c91 is described below
commit be5dc80c9182244c81990ffcd221cc2a12d9f82a
Author: Christofer Dutz <[email protected]>
AuthorDate: Sun Jun 28 13:35:52 2026 +0200
fix: Fixed a potential race condition in the connection-cache.
---
.../utils/cache/CachedPlcConnectionManager.java | 18 +++++++++-
.../java/utils/cache/ConnectionContainerTest.java | 40 ++++++++++------------
2 files changed, 36 insertions(+), 22 deletions(-)
diff --git
a/plc4j/tools/connection-cache/src/main/java/org/apache/plc4x/java/utils/cache/CachedPlcConnectionManager.java
b/plc4j/tools/connection-cache/src/main/java/org/apache/plc4x/java/utils/cache/CachedPlcConnectionManager.java
index 535d93d048..b3f9ba5575 100644
---
a/plc4j/tools/connection-cache/src/main/java/org/apache/plc4x/java/utils/cache/CachedPlcConnectionManager.java
+++
b/plc4j/tools/connection-cache/src/main/java/org/apache/plc4x/java/utils/cache/CachedPlcConnectionManager.java
@@ -85,6 +85,15 @@ public class CachedPlcConnectionManager implements
PlcConnectionManager, AutoClo
private static final long DEFAULT_IDLE_PING_THRESHOLD_MS =
TimeUnit.SECONDS.toMillis(30);
private static final long DEFAULT_CLOSE_TIMEOUT_MS =
TimeUnit.SECONDS.toMillis(5);
+ /**
+ * Grace margin added to the caller-facing lease {@code get()} timeout, on
top of the per-connection
+ * max-wait. It ensures the container's own wait timeout (scheduled at
exactly max-wait) fires and
+ * marks a queued waiter "done" before this manager gives up — preventing
a returned connection from
+ * being handed to a waiter whose caller has already timed out. Purely an
upper safety bound; the
+ * {@code get()} returns as soon as the future completes.
+ */
+ private static final long LEASE_WAIT_GRACE_MS =
TimeUnit.SECONDS.toMillis(1);
+
private final PlcConnectionManager connectionManager;
private final ScheduledExecutorService scheduler;
private final long maxIdleTimeMs;
@@ -155,7 +164,14 @@ public class CachedPlcConnectionManager implements
PlcConnectionManager, AutoClo
// Lease the connection - THIS WILL BLOCK IF ALREADY LEASED
Future<PlcConnection> leaseFuture = container.lease();
try {
- return leaseFuture.get(maxWaitTimeMs, TimeUnit.MILLISECONDS);
+ // Wait a grace margin BEYOND the container's own max-wait
timeout. The container already
+ // guarantees the lease future completes within maxWaitTimeMs (it
schedules its own wait
+ // timeout), so letting our get() time out at exactly
maxWaitTimeMs would race that: if our
+ // get() won, we'd abandon a queue entry that is still "not done",
and a concurrent return
+ // could then hand the connection to that phantom waiter — wedging
the connection. The grace
+ // lets the container's timeout win and mark the entry done first;
get() still returns the
+ // instant the future completes, so a normal wait-timeout is not
slowed down.
+ return leaseFuture.get(maxWaitTimeMs + LEASE_WAIT_GRACE_MS,
TimeUnit.MILLISECONDS);
} catch (ExecutionException | InterruptedException | TimeoutException
e) {
throw new PlcConnectionException("Error acquiring lease for
connection", e);
}
diff --git
a/plc4j/tools/connection-cache/src/test/java/org/apache/plc4x/java/utils/cache/ConnectionContainerTest.java
b/plc4j/tools/connection-cache/src/test/java/org/apache/plc4x/java/utils/cache/ConnectionContainerTest.java
index 23951adbc1..6752128531 100644
---
a/plc4j/tools/connection-cache/src/test/java/org/apache/plc4x/java/utils/cache/ConnectionContainerTest.java
+++
b/plc4j/tools/connection-cache/src/test/java/org/apache/plc4x/java/utils/cache/ConnectionContainerTest.java
@@ -26,6 +26,7 @@ import org.junit.jupiter.api.Test;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
+import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
@@ -34,6 +35,7 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.Mockito.*;
@@ -220,20 +222,17 @@ class ConnectionContainerTest {
100 // Very short idle ping threshold
);
- // Act - Lease, return, wait for validation threshold
+ // Act - Lease, return, then re-lease once the connection is due for
validation.
PlcConnection leased1 = container.lease().get();
leased1.close();
- Thread.sleep(200); // Wait past idle ping threshold
-
- // Second lease should trigger validation
- PlcConnection leased2 = container.lease().get();
-
- // Assert
- verify(mockConnection, atLeastOnce()).ping();
-
- // Cleanup
- leased2.close();
+ // The container arms validation via a scheduled task
~idlePingThreshold after the return.
+ // That scheduler tick can lag under load, so poll the lease/validate
path until the idle
+ // connection is actually pinged instead of assuming a fixed
wall-clock delay.
+ await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
+ container.lease().get().close();
+ verify(mockConnection, atLeastOnce()).ping();
+ });
}
@Test
@@ -261,22 +260,21 @@ class ConnectionContainerTest {
// Make first connection's ping fail
when(mockConnection.ping()).thenReturn(CompletableFuture.failedFuture(new
RuntimeException("Ping failed")));
- // Act - Lease, return, wait for validation threshold
+ // Act - Lease, return, then re-lease once the connection is due for
validation.
PlcConnection leased1 = container.lease().get();
leased1.close();
- Thread.sleep(200); // Wait past idle ping threshold
-
- // Second lease should fail validation and create new connection
- PlcConnection leased2 = container.lease().get();
+ // Validation is armed by a scheduled task ~idlePingThreshold after
the return, which can
+ // lag under load. Poll the lease/validate path until the (failing)
ping has run; once the
+ // first connection's ping fails the container discards it and builds
a replacement.
+ await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
+ container.lease().get().close();
+ verify(mockConnection, atLeastOnce()).ping();
+ });
- // Assert - Should have created connection twice
+ // Assert - the failed ping replaced the first connection exactly once.
assertEquals(2, callCount[0]);
- verify(mockConnection, atLeastOnce()).ping();
verify(mockConnection, times(1)).close(); // Old connection closed
-
- // Cleanup
- leased2.close();
}
@Test
