feat(core): handle deserialization error for WorkerTrigger

kestra-io · Dec 5, 2023 · f5a7df1 · f5a7df1
1 parent af79920
commit f5a7df1
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 5 deletions.
diff --git a/core/src/main/java/io/kestra/core/runners/Worker.java b/core/src/main/java/io/kestra/core/runners/Worker.java
@@ -19,6 +19,7 @@
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.tasks.retrys.AbstractRetry;
 import io.kestra.core.models.triggers.PollingTriggerInterface;
+import io.kestra.core.models.triggers.TriggerContext;
 import io.kestra.core.queues.QueueException;
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
@@ -156,8 +157,12 @@ private void handleDeserializationError(DeserializationException deserialization
                     // try to deserialize the taskRun to fail it
                     var taskRun = MAPPER.treeToValue(json.get("taskRun"), TaskRun.class);
                     this.workerTaskResultQueue.emit(new WorkerTaskResult(taskRun.fail()));
+                } else if ("trigger".equals(type)) {
+                    // try to deserialize the triggerContext to fail it
+                    var triggerContext = MAPPER.treeToValue(json.get("triggerContext"), TriggerContext.class);
+                    var workerTriggerResult = WorkerTriggerResult.builder().triggerContext(triggerContext).success(false).execution(Optional.empty()).build();
+                    this.workerTriggerResultQueue.emit(workerTriggerResult);
                 }
-                // TODO for now we cannot handle deserialization error for a WorkerTrigger as we cannot create a valid WorkerTriggerResult as it contains the trigger itself
             }
             catch (IOException e) {
                 // silently ignore the message if we cannot do anything about it

diff --git a/core/src/test/java/io/kestra/core/runners/DeserializationIssuesCaseTest.java b/core/src/test/java/io/kestra/core/runners/DeserializationIssuesCaseTest.java
@@ -184,7 +184,7 @@ public void workerTaskDeserializationIssue(Consumer<QueueMessage> sendToQueue) t
         assertThat(workerTaskResult.get().getTaskRun().getState().getCurrent(), is(State.Type.FAILED));
     }
 
-    public void workerTriggerDeserializationIssue(Consumer<QueueMessage> sendToQueue) throws InterruptedException {
+    public void workerTriggerDeserializationIssue(Consumer<QueueMessage> sendToQueue) throws TimeoutException {
         AtomicReference<WorkerTriggerResult> workerTriggerResult = new AtomicReference<>();
         workerTriggerResultQueue.receive(either -> {
             if (either != null) {
@@ -194,9 +194,12 @@ public void workerTriggerDeserializationIssue(Consumer<QueueMessage> sendToQueue
 
         sendToQueue.accept(new QueueMessage(WorkerJob.class, INVALID_WORKER_TRIGGER_KEY, INVALID_WORKER_TRIGGER_VALUE));
 
-        // Invalid worker trigger will be ignored, so we just check that no messages are received
-        Thread.sleep(500);
-        assertThat(workerTriggerResult.get(), nullValue());
+        Await.until(
+            () -> workerTriggerResult.get() != null,
+            Duration.ofMillis(100),
+            Duration.ofMinutes(1)
+        );
+        assertThat(workerTriggerResult.get().getSuccess(), is(Boolean.FALSE));
     }
 
     public void flowDeserializationIssue(Consumer<QueueMessage> sendToQueue) throws TimeoutException {