diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SenderConnectionDispatcherTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SenderConnectionDispatcherTest.java index e7c0d7c1..71f060ae 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SenderConnectionDispatcherTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SenderConnectionDispatcherTest.java @@ -100,10 +100,12 @@ public void testFullInboxDropsOldestAndCounts() throws Exception { // entry and admit the new one. Later connection events carry the // freshest state, so dropping the head loses the least information. CountDownLatch unblock = new CountDownLatch(1); + CountDownLatch listenerEntered = new CountDownLatch(1); List received = new ArrayList<>(); Object lock = new Object(); CountDownLatch allDelivered = new CountDownLatch(5); try (SenderConnectionDispatcher d = new SenderConnectionDispatcher(ev -> { + listenerEntered.countDown(); try { unblock.await(); } catch (InterruptedException ignored) { @@ -114,11 +116,12 @@ public void testFullInboxDropsOldestAndCounts() throws Exception { } allDelivered.countDown(); }, /*capacity=*/ 4)) { - // First offer starts the dispatcher and lands in the listener - // immediately (and blocks there), freeing one slot. Now we can - // fill the bounded inbox to capacity (4), then overflow. + // First offer starts the dispatcher. Wait until it lands in the + // listener and blocks there before filling the bounded inbox and + // overflowing it. Assert.assertTrue(d.offer(buildEvent(0))); - TimeUnit.MILLISECONDS.sleep(50); + Assert.assertTrue("dispatcher should take the head into the listener within 5s", + listenerEntered.await(5, TimeUnit.SECONDS)); for (int i = 1; i <= 4; i++) { Assert.assertTrue("inbox should accept offer " + i, d.offer(buildEvent(i))); diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SenderErrorDispatcherTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SenderErrorDispatcherTest.java index 957e36d4..69725fb7 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SenderErrorDispatcherTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SenderErrorDispatcherTest.java @@ -102,10 +102,12 @@ public void testFullInboxDropsOldestAndCounts() throws Exception { // and admit the new one. The latest entry is always the most // informative, so the FIFO head loses, not the new arrival. CountDownLatch unblock = new CountDownLatch(1); + CountDownLatch handlerEntered = new CountDownLatch(1); List received = new ArrayList<>(); Object lock = new Object(); CountDownLatch allDelivered = new CountDownLatch(5); try (SenderErrorDispatcher d = new SenderErrorDispatcher(err -> { + handlerEntered.countDown(); try { unblock.await(); } catch (InterruptedException ignored) { @@ -120,9 +122,11 @@ public void testFullInboxDropsOldestAndCounts() throws Exception { // immediately (and blocks there). Now we can fill the bounded // inbox to capacity, then overflow. Assert.assertTrue(d.offer(buildError(0))); - // Give the dispatcher a moment to take the head into the - // handler so subsequent offers don't get an extra slot. - TimeUnit.MILLISECONDS.sleep(50); + // The dispatcher polls the head off the inbox before invoking + // the handler, so once the handler is entered the head's slot + // is free and subsequent offers cannot get an extra slot. + Assert.assertTrue("dispatcher should take the head into the handler within 5s", + handlerEntered.await(5, TimeUnit.SECONDS)); for (int i = 1; i <= 4; i++) { Assert.assertTrue("inbox should accept offer " + i, d.offer(buildError(i)));