fix(tests): added debug information in case of WorkerTest.failOnWorke…

…rTaskWithFlowable test failure
kestra-io · Dec 22, 2023 · c985c31 · c985c31
1 parent 3f8285a
commit c985c31
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 2 deletions.
diff --git a/core/src/main/java/io/kestra/core/utils/Await.java b/core/src/main/java/io/kestra/core/utils/Await.java
@@ -28,14 +28,22 @@ public static void until(BooleanSupplier condition, Duration sleep) {
     }
 
     public static void until(BooleanSupplier condition, Duration sleep, Duration timeout) throws TimeoutException {
+        until(null, condition, sleep, timeout);
+    }
+
+    public static void until(Supplier<String> errorMessageInCaseOfFailure, BooleanSupplier condition, Duration sleep, Duration timeout) throws TimeoutException {
         if (sleep == null) {
             sleep = defaultSleep;
         }
 
         long start = System.currentTimeMillis();
         while (!condition.getAsBoolean()) {
             if (System.currentTimeMillis() - start > timeout.toMillis()) {
-                throw new TimeoutException(String.format("Await failed to terminate within %s", timeout));
+                throw new TimeoutException(String.format(
+                    "Await failed to terminate within %s.%s",
+                    timeout,
+                    errorMessageInCaseOfFailure == null ? "" : " " + errorMessageInCaseOfFailure.get()
+                ));
             } else {
                 try {
                     Thread.sleep(sleep.toMillis());

diff --git a/core/src/test/java/io/kestra/core/runners/WorkerTest.java b/core/src/test/java/io/kestra/core/runners/WorkerTest.java
@@ -1,5 +1,6 @@
 package io.kestra.core.runners;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.google.common.collect.ImmutableMap;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.ExecutionKilled;
@@ -10,6 +11,7 @@
 import io.kestra.core.models.tasks.ResolvedTask;
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
+import io.kestra.core.serializers.JacksonMapper;
 import io.kestra.core.tasks.flows.Pause;
 import io.kestra.core.tasks.flows.WorkingDirectory;
 import io.kestra.core.tasks.test.Sleep;
@@ -30,6 +32,7 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
 
+import static io.kestra.core.utils.Rethrow.throwSupplier;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.nullValue;
@@ -84,7 +87,7 @@ void workerGroup() {
     }
 
     @Test
-    void failOnWorkerTaskWithFlowable() throws TimeoutException {
+    void failOnWorkerTaskWithFlowable() throws TimeoutException, JsonProcessingException {
         Worker worker = new Worker(applicationContext, 8, null);
         worker.run();
 
@@ -122,6 +125,10 @@ void failOnWorkerTaskWithFlowable() throws TimeoutException {
         workerTaskQueue.emit(workerTask);
 
         Await.until(
+            throwSupplier(() -> {
+                WorkerTaskResult taskResult = workerTaskResult.get();
+                return "WorkerTaskResult was " + (taskResult == null ? null : JacksonMapper.ofJson().writeValueAsString(taskResult));
+            }),
             () -> workerTaskResult.get() != null && workerTaskResult.get().getTaskRun().getState().isFailed(),
             Duration.ofMillis(100),
             Duration.ofMinutes(1)