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();