fix(core): multiple condition test is flaky

kestra-io · Nov 13, 2024 · dcab506 · dcab506
1 parent f0843ed
commit dcab506
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 16 deletions.
diff --git a/core/src/test/java/io/kestra/core/runners/MultipleConditionTriggerCaseTest.java b/core/src/test/java/io/kestra/core/runners/MultipleConditionTriggerCaseTest.java
@@ -24,8 +24,7 @@
 import reactor.core.publisher.Flux;
 
 import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.*;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 @Singleton
@@ -95,18 +94,13 @@ public void trigger() throws InterruptedException, TimeoutException, QueueExcept
     }
 
     public void failed() throws InterruptedException, TimeoutException, QueueException {
-        CountDownLatch countDownLatch = new CountDownLatch(2);
-        ConcurrentHashMap<String, Execution> ended = new ConcurrentHashMap<>();
-
+        CountDownLatch countDownLatch = new CountDownLatch(1);
+        AtomicReference<Execution> listener = new AtomicReference<>();
         Flux<Execution> receive = TestsUtils.receive(executionQueue, either -> {
-            synchronized (ended) {
-                Execution execution = either.getLeft();
-                if (execution.getState().getCurrent().isTerminated() && !execution.getFlowId().equals("trigger-flow-listener-namespace-condition")) {
-                    if (!ended.containsKey(execution.getId())) {
-                        ended.put(execution.getId(), execution);
-                        countDownLatch.countDown();
-                    }
-                }
+            Execution execution = either.getLeft();
+            if (execution.getFlowId().equals("trigger-flow-listener-namespace-condition") && execution.getState().getCurrent().isTerminated() ) {
+                countDownLatch.countDown();
+                listener.set(execution);
             }
         });
 
@@ -117,7 +111,7 @@ public void failed() throws InterruptedException, TimeoutException, QueueExcepti
 
         // wait a little to be sure that the trigger is not launching execution
         Thread.sleep(1000);
-        assertThat(ended.size(), is(1));
+        assertThat(listener.get(), nullValue());
 
         // second one
         execution = runnerUtils.runOne(null, "io.kestra.tests.trigger", "trigger-multiplecondition-flow-d", Duration.ofSeconds(60));
@@ -127,7 +121,8 @@ public void failed() throws InterruptedException, TimeoutException, QueueExcepti
         // trigger was not done
         assertTrue(countDownLatch.await(10, TimeUnit.SECONDS));
         receive.blockLast();
-        assertThat(ended.size(), is(2));
+        assertThat(listener.get(), notNullValue());
+        assertThat(listener.get().getState().getCurrent(), is(State.Type.SUCCESS));
     }
 
     public void basicExecutionsCondition() throws InterruptedException, TimeoutException, QueueException {

diff --git a/core/src/test/resources/flows/valids/trigger-flow-listener-namespace-condition.yaml b/core/src/test/resources/flows/valids/trigger-flow-listener-namespace-condition.yaml
@@ -15,4 +15,7 @@ triggers:
     type: io.kestra.plugin.core.trigger.Flow
     conditions:
       - type: io.kestra.plugin.core.condition.ExecutionNamespaceCondition
-        namespace: io.kestra.tests.trigger
+        namespace: io.kestra.tests.trigger
+      - type: io.kestra.plugin.core.condition.ExecutionStatusCondition
+        in:
+          - SUCCESS