fix(kafka-runner): remove a plugins can lead to stop of executor

related to #608
kestra-io · Sep 21, 2022 · 2e0e6ce · 2e0e6ce
1 parent 8372709
commit 2e0e6ce
Show file tree

Hide file tree

Showing 8 changed files with 45 additions and 12 deletions.
diff --git a/core/src/main/java/io/kestra/core/runners/DefaultFlowExecutor.java b/core/src/main/java/io/kestra/core/runners/DefaultFlowExecutor.java
@@ -42,4 +42,9 @@ public Optional<Flow> findById(String namespace, String id, Optional<Integer> re
             return flowRepositoryInterface.findById(namespace, id, revision);
         }
     }
+
+    @Override
+    public Boolean isReady() {
+        return true;
+    }
 }
diff --git a/core/src/main/java/io/kestra/core/runners/Executor.java b/core/src/main/java/io/kestra/core/runners/Executor.java
@@ -36,6 +36,10 @@ public Executor(WorkerTaskResult workerTaskResult) {
         this.joined = workerTaskResult;
     }
 
+    public Boolean canBeProcessed() {
+        return !(this.getException() != null || this.getFlow() == null || this.getExecution().isDeleted());
+    }
+
     public Executor withFlow(Flow flow) {
         this.flow = flow;
 

diff --git a/core/src/main/java/io/kestra/core/runners/ExecutorService.java b/core/src/main/java/io/kestra/core/runners/ExecutorService.java
@@ -56,7 +56,7 @@ protected FlowExecutorInterface flowExecutorInterface() {
 
     public Executor process(Executor executor) {
         // previous failed (flow join can fail), just forward
-        if (executor.getException() != null || executor.getExecution().isDeleted()) {
+        if (!executor.canBeProcessed()) {
             return executor;
         }
 

diff --git a/core/src/main/java/io/kestra/core/runners/FlowExecutorInterface.java b/core/src/main/java/io/kestra/core/runners/FlowExecutorInterface.java
@@ -11,6 +11,8 @@ public interface FlowExecutorInterface {
 
     Optional<Flow> findById(String namespace, String id, Optional<Integer> revision);
 
+    Boolean isReady();
+
     default Optional<Flow> findByIdFromFlowTask(String namespace, String id, Optional<Integer> revision, String fromNamespace, String fromId) {
         return this.findById(
             namespace,

diff --git a/core/src/main/java/io/kestra/core/services/ConditionService.java b/core/src/main/java/io/kestra/core/services/ConditionService.java
@@ -126,7 +126,7 @@ public boolean isTerminatedWithListeners(Flow flow, Execution execution) {
     }
 
     public List<ResolvedTask> findValidListeners(Flow flow, Execution execution) {
-        if (flow.getListeners() == null) {
+        if (flow == null || flow.getListeners() == null) {
             return new ArrayList<>();
         }
 

diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/KafkaFlowExecutor.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/KafkaFlowExecutor.java
@@ -39,7 +39,7 @@ public synchronized void setStore(SafeKeyValueStore<String, Flow> store) {
     @SneakyThrows
     private void await() {
         if (flowsLast == null || store == null) {
-            Await.until(() -> this.flowsLast != null && store != null, Duration.ofMillis(100), Duration.ofMinutes(5));
+            Await.until(() -> this.isReady() == true, Duration.ofMillis(100), Duration.ofMinutes(5));
         }
     }
 
@@ -67,4 +67,9 @@ public Optional<Flow> findById(String namespace, String id, Optional<Integer> re
 
         return this.store.get(Flow.uid(namespace, id, revision));
     }
+
+    @Override
+    public Boolean isReady() {
+        return this.flowsLast != null && store != null;
+    }
 }
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/services/KafkaStreamSourceService.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/services/KafkaStreamSourceService.java
@@ -22,7 +22,6 @@
 
 import java.time.Duration;
 import java.util.concurrent.TimeoutException;
-import java.util.function.Supplier;
 
 @Singleton
 @Slf4j
@@ -74,16 +73,29 @@ public Executor joinFlow(Executor executor, Boolean withDefaults) {
 
         try {
             // pooling of new flow can be delayed on ExecutorStore, we maybe need to wait that the flow is updated
-            flow = Await.until(
-                () -> flowExecutorInterface.findByExecution(executor.getExecution()).orElse(null),
-                Duration.ofMillis(100),
-                Duration.ofMinutes(5)
-            );
+            if (!flowExecutorInterface.isReady()) {
+                flow = Await.until(
+                    () -> flowExecutorInterface.findByExecution(executor.getExecution()).orElse(null),
+                    Duration.ofMillis(100),
+                    Duration.ofMinutes(5)
+                );
+            } else {
+                flow = flowExecutorInterface.findByExecution(executor.getExecution())
+                    .orElseThrow(() -> new TimeoutException("Unable to find flow with flow executor ready"));
+            }
         } catch (TimeoutException e) {
+            // execution is failed, can't find flow, avoid recursive exception, skipped it.
+            if (executor.getExecution().getState().isFailed()) {
+                return executor;
+            }
+
             return executor.withException(
-                new Exception("Unable to find flow with namespace: '" + executor.getExecution().getNamespace() + "'" +
-                    ", id: '" + executor.getExecution().getFlowId() + "', " +
-                    "revision '" + executor.getExecution().getFlowRevision() + "'"),
+                new Exception(
+                    "Unable to find flow with namespace: '" + executor.getExecution().getNamespace() + "'" +
+                        ", id: '" + executor.getExecution().getFlowId() + "', " +
+                        "revision '" + executor.getExecution().getFlowRevision() + "'",
+                    e
+                ),
                 "joinFlow"
             );
         }

diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/ExecutorNextTransformer.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/ExecutorNextTransformer.java
@@ -33,6 +33,11 @@ public void init(final ProcessorContext context) {
 
     @Override
     public Executor transform(final String key, final Executor value) {
+        // previous failed (flow join can fail), just forward
+        if (!value.canBeProcessed()) {
+            return value;
+        }
+
         Executor executor = executorService.process(value);
 
         if (executor.getNexts().size() == 0) {