More complicated latching, now with extra comments

dakrone · dakrone · commit b4da58a26c46 · 2026-03-09T14:51:28.000-06:00
diff --git a/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java b/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java
@@ -154,6 +154,7 @@ public class DataStreamLifecycleServiceTests extends ESTestCase {
     private List<TransportRequest> clientSeenRequests;
     private DoExecuteDelegate clientDelegate;
     private volatile CountDownLatch clientWaitLatch;
+    private volatile CountDownLatch invokerWaitLatch;
     private ClusterService clusterService;
     private final DataStreamGlobalRetentionSettings globalRetentionSettings = DataStreamGlobalRetentionSettings.create(
         ClusterSettings.createBuiltInClusterSettings()
@@ -206,6 +207,7 @@ public void setupServices() {
             actions
         );
         clientWaitLatch = null;
+        invokerWaitLatch = null;
         clientDelegate = null;
     }
 
@@ -288,6 +290,7 @@ public void testDLMRunsOnlyOnce() {
         ClusterState state = ClusterState.builder(ClusterName.DEFAULT).putProjectMetadata(builder).build();
 
         clientWaitLatch = new CountDownLatch(1);
+        invokerWaitLatch = new CountDownLatch(1);
         AtomicBoolean runCompleted = new AtomicBoolean(false);
         // Should block because of the latch
         Thread t = new Thread(() -> {
@@ -296,11 +299,23 @@ public void testDLMRunsOnlyOnce() {
         });
         t.start();
 
+        // So it's possible for the thread to be started above, but for the
+        // actual `.run` invocation not to have been called by this point.
+        // What we actually need to do is wait for some moment where we know
+        // we're in the middle of the DLM service. In order to do that, we wait
+        // for the "invokerWaitLatch" which is counted down inside of the fake
+        // client. That way we know that the DLM service is running, but is
+        // "paused" because of the `clientWaitLatch`.
+        try {
+            assertTrue("expected the client to count the latch down, but it didn't", invokerWaitLatch.await(10, TimeUnit.SECONDS));
+        } catch (InterruptedException e) {
+            fail("expected the client to have been invoked, but it never was");
+        }
         // Will return immediately because it's already running
         logger.info("--> second 'run' invocation");
         dataStreamLifecycleService.run(state);
 
-        // Let the first invocation proceed
+        // Let the first invocation proceed by decrementing clientWatchLatch.
         logger.info("--> decrementing latch");
         clientWaitLatch.countDown();
         try {
@@ -309,6 +324,8 @@ public void testDLMRunsOnlyOnce() {
         } catch (InterruptedException e) {
             throw new ElasticsearchException(e);
         }
+        // Always check that we finished the initial `.run` call that we did
+        // inside the thread.
         assertTrue(runCompleted.get());
     }
 
@@ -1874,6 +1891,9 @@ protected <Request extends ActionRequest, Response extends ActionResponse> void
                 if (clientDelegate != null) {
                     clientDelegate.doExecute(action, request, listener);
                 }
+                if (invokerWaitLatch != null) {
+                    invokerWaitLatch.countDown();
+                }
                 if (clientWaitLatch != null && clientWaitLatch.getCount() > 0) {
                     try {
                         logger.info("--> blocking client invocation");
