From b62bf91124e2ec1357966df0b1ae4e4b53ecb2f2 Mon Sep 17 00:00:00 2001 From: Peter Isberg Date: Thu, 5 Nov 2026 18:54:14 +0100 Subject: [PATCH] test(telemetry): wait for a dropped event before the starved drainer runs (#1152) testMpscConcurrencyWithAStarvedConsumer started the drainer after a fixed 1,300 ms and asserted a drop. A drop needs the buffer full for 1 s with no consumer progress, so the assertion held only if the 8 producers filled 1,024 slots within about 300 ms of the start. Under the full suite's parallel forks it failed once (2026-10-31). The starved case now starts the drainer once buffer.droppedCount() > 0, with a 10 s cap that fails the test. The accounting assertions are unchanged. Verified: the original test fails with the issue's message when producers sleep 1.5 s before publishing (AssertionFailedError: A starved consumer should have cost a drop, expected true but was false). The fixed test passes 6/6 unmodified and passes with the same injected delay. Co-Authored-By: Claude Haiku 5.5 --- .../telemetry/TelemetryEventBufferTest.java | 31 +++++++++++++------ 1 file changed, 22 insertions(+), 9 deletions(-) diff --git a/async-test-lib/src/test/java/se/deversity/asynctest/telemetry/TelemetryEventBufferTest.java b/async-test-lib/src/test/java/se/deversity/asynctest/telemetry/TelemetryEventBufferTest.java index 7f9605f7f..75e01246a 100644 --- a/async-test-lib/src/test/java/se/deversity/asynctest/telemetry/TelemetryEventBufferTest.java +++ b/async-test-lib/src/test/java/se/deversity/asynctest/telemetry/TelemetryEventBufferTest.java @@ -97,8 +97,8 @@ void testMpscConcurrency() throws Exception { @Test void testMpscConcurrencyWithAStarvedConsumer() throws Exception { - // The drain thread starts after the publish give-up bound (1 s), which is what a - // full-suite run does to a scheduled drainer now and then (#987). + // The drain thread starts only after the publish give-up bound (1 s) has dropped an + // event, which is what a full-suite run does to a scheduled drainer now and then (#987). runMpsc(1_300); } @@ -112,13 +112,6 @@ private void runMpsc(long drainDelayMillis) throws Exception { List drainedEvents = Collections.synchronizedList(new ArrayList<>()); ScheduledExecutorService drainExecutor = Executors.newSingleThreadScheduledExecutor(); - // Run consumer in background to drain buffer constantly. - drainExecutor.scheduleAtFixedRate(() -> { - buffer.drain((tid, field, write) -> { - drainedEvents.add(tid + ":" + field); - }); - }, drainDelayMillis, 10, TimeUnit.MILLISECONDS); - CountDownLatch latch = new CountDownLatch(producerThreadsCount); for (int p = 0; p < producerThreadsCount; p++) { final long threadId = p + 100; @@ -133,6 +126,26 @@ private void runMpsc(long drainDelayMillis) throws Exception { }); } + if (drainDelayMillis > 0) { + // A starved consumer: the drainer arrives only after the producers have filled the + // buffer and the publish give-up has dropped an event. Waiting on that state, not on + // a wall-clock delay, keeps the test from depending on how fast the pool starts (#1152). + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while (buffer.droppedCount() == 0) { + if (System.nanoTime() > deadline) { + fail("No event was dropped within 10 s of a full buffer with no consumer"); + } + Thread.sleep(10); + } + } + + // Run consumer in background to drain buffer constantly. + drainExecutor.scheduleAtFixedRate(() -> { + buffer.drain((tid, field, write) -> { + drainedEvents.add(tid + ":" + field); + }); + }, 0, 10, TimeUnit.MILLISECONDS); + assertTrue(latch.await(5, TimeUnit.SECONDS), "Producers did not finish publishing in time"); executor.shutdown();