fix: use onInvalidSyntax properly when invalid flows

Author: François Delbrayelle

src/main/java/io/kestra/plugin/git/NamespaceSync.java

@@ -234,7 +234,7 @@ public Output run(RunContext runContext) throws Exception {
         }
 
         // Fetch Kestra state limited to target namespace only
-        KestraState kestraState = loadKestraState(runContext, rNamespace);
+        KestraState kestraState = loadKestraState(runContext, rNamespace, rOnInvalidSyntax);
         Map<String, byte[]> namespaceFiles = listNamespaceFiles(runContext, rNamespace);
 
         List<DiffLine> diffs = new ArrayList<>();
@@ -323,7 +323,7 @@ private void planFlows(RunContext rc, Path baseDir, GitTree gitTree, KestraState
                                 throw new FlowProcessingException(flowValidated.getConstraints());
                             }
                             fs.importFlow(tenant, gitNode.rawYaml, false);
-                        } catch (FlowProcessingException e) {
+                        } catch (Exception e) {
                             handleInvalid(rc, rInvalid, "FLOW " + key, e);
                         }
                     });
@@ -383,7 +383,7 @@ private void planFlows(RunContext rc, Path baseDir, GitTree gitTree, KestraState
                             }
 
                             fs.importFlow(tenant, gitNode.rawYaml, false);
-                        } catch (FlowProcessingException e) {
+                        } catch (Exception e) {
                             handleInvalid(rc, rInvalid, "FLOW " + key, e);
                         }
                     });
@@ -478,11 +478,19 @@ private static class GitTree {
     private record KestraState(Map<String, FlowWithSource> flows) {
     }
 
-    private KestraState loadKestraState(RunContext rc, String rootNamespace) {
+    private KestraState loadKestraState(RunContext rc, String rootNamespace, OnInvalidSyntax rOnInvalidSyntax) {
         FlowService fs = flowService(rc);
         String tenant = rc.flowInfo().tenantId();
 
-        Map<String, FlowWithSource> flowsWithSource = fs.findByNamespaceWithSource(tenant, rootNamespace).stream()
+        List<FlowWithSource> allFlows;
+        try {
+            allFlows = fs.findByNamespaceWithSource(tenant, rootNamespace);
+        } catch (Exception e) {
+            handleInvalid(rc, rOnInvalidSyntax, "flows for namespace " + rootNamespace, e);
+            allFlows = List.of();
+        }
+
+        Map<String, FlowWithSource> flowsWithSource = allFlows.stream()
             .collect(Collectors.toMap(f -> key(f.getNamespace(), f.getId()), Function.identity(), (a, b) -> a));
 
         return new KestraState(flowsWithSource);

src/main/java/io/kestra/plugin/git/TenantSync.java

@@ -9,6 +9,7 @@
 import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.RunnableTask;
 import io.kestra.core.runners.RunContext;
+import io.kestra.core.serializers.YamlParser;
 import io.kestra.plugin.git.services.GitService;
 import io.kestra.sdk.KestraClient;
 import io.kestra.sdk.api.FilesApi;
@@ -34,6 +35,8 @@
 import java.nio.file.Path;
 import java.util.*;
 import java.util.stream.Collectors;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipInputStream;
 
 import static org.eclipse.jgit.transport.RemoteRefUpdate.Status.*;
 
@@ -242,7 +245,7 @@ public Output run(RunContext runContext) throws Exception {
                 }
             }
 
-            List<FlowWithSource> kestraFlows = fetchFlowsFromKestra(kestraClient, runContext, namespace);
+            List<FlowWithSource> kestraFlows = fetchFlowsFromKestra(kestraClient, runContext, namespace, rOnInvalidSyntax);
             Map<String, byte[]> kestraFiles = listNamespaceFiles(kestraClient, runContext, namespace);
 
             planNamespace(
@@ -728,7 +731,7 @@ private void planDashboards(
         }
     }
 
-    private List<FlowWithSource> fetchFlowsFromKestra(KestraClient kestraClient, RunContext runContext, String namespace) {
+    private List<FlowWithSource> fetchFlowsFromKestra(KestraClient kestraClient, RunContext runContext, String namespace, OnInvalidSyntax rOnInvalidSyntax) {
         try {
             // Export all flows from Kestra for the given namespace (including sub-namespaces)
             byte[] zippedFlows = kestraClient.flows().exportFlowsByQuery(
@@ -740,17 +743,24 @@ private List<FlowWithSource> fetchFlowsFromKestra(KestraClient kestraClient, Run
 
             try (
                 var bais = new ByteArrayInputStream(zippedFlows);
-                var zis = new java.util.zip.ZipInputStream(bais)
+                var zis = new ZipInputStream(bais)
             ) {
-                java.util.zip.ZipEntry entry;
+                ZipEntry entry;
                 while ((entry = zis.getNextEntry()) != null) {
                     if (!entry.getName().endsWith(".yml") && !entry.getName().endsWith(".yaml")) {
                         continue;
                     }
 
-                    String yaml = new String(zis.readAllBytes(), StandardCharsets.UTF_8);
+                    var entryName = entry.getName();
+                    var yaml = new String(zis.readAllBytes(), StandardCharsets.UTF_8);
 
-                    io.kestra.core.models.flows.Flow parsed = io.kestra.core.serializers.YamlParser.parse(yaml, io.kestra.core.models.flows.Flow.class);
+                    io.kestra.core.models.flows.Flow parsed;
+                    try {
+                        parsed = YamlParser.parse(yaml, io.kestra.core.models.flows.Flow.class);
+                    } catch (Exception e) {
+                        handleInvalid(runContext, rOnInvalidSyntax, "FLOW from entry " + entryName, e);
+                        continue;
+                    }
 
                     if (!namespace.equals(parsed.getNamespace())) {
                         continue;
@@ -762,6 +772,9 @@ private List<FlowWithSource> fetchFlowsFromKestra(KestraClient kestraClient, Run
 
             return flows;
 
+        } catch (KestraRuntimeException e) {
+            // Re-throw KestraRuntimeException from handleInvalid(FAIL) without wrapping
+            throw e;
         } catch (Exception e) {
             throw new KestraRuntimeException("Failed to export flows from Kestra for namespace " + namespace, e);
         }