test(telemetry): stop the starved-consumer case racing the pool start (#1152)

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