Repository: kestra-io/kestra
Issue: Flow listeners should be fault tolerant on invalid flow & plugins
Labels: bug

```
io.micronaut.context.exceptions.BeanInstantiationException: Error instantiating bean of type  [io.kestra.jdbc.runner.JdbcExecutor]

Message: io/reactivex/functions/BiConsumer
Path Taken: new JdbcExecutor() --> JdbcExecutor.flowListeners
	at io.micronaut.context.DefaultBeanContext.resolveByBeanFactory(DefaultBeanContext.java:2324)
	at io.micronaut.context.DefaultBeanContext.doCreateBean(DefaultBeanContext.java:2279)
	at io.micronaut.context.DefaultBeanContext.doCreateBean(DefaultBeanContext.java:2291)
	at io.micronaut.context.DefaultBeanContext.createRegistration(DefaultBeanContext.java:3054)
	at io.micronaut.context.SingletonScope.getOrCreate(SingletonScope.java:80)
	at io.micronaut.context.DefaultBeanContext.findOrCreateSingletonBeanRegistration(DefaultBeanContext.java:2956)
	at io.micronaut.context.DefaultBeanContext.resolveBeanRegistration(DefaultBeanContext.java:2917)
	at io.micronaut.context.DefaultBeanContext.resolveBeanRegistration(DefaultBeanContext.java:2730)
	at io.micronaut.context.DefaultBeanContext.getBean(DefaultBeanContext.java:1729)
	at io.micronaut.context.AbstractBeanResolutionContext.getBean(AbstractBeanResolutionContext.java:89)
	at io.micronaut.context.AbstractInitializableBeanDefinition.resolveBean(AbstractInitializableBeanDefinition.java:2165)
	at io.micronaut.context.AbstractInitializableBeanDefinition.getBeanForField(AbstractInitializableBeanDefinition.java:1672)
	at io.kestra.jdbc.runner.$JdbcExecutor$Definition.inject(Unknown Source)
	at io.kestra.jdbc.runner.$JdbcExecutor$Definition.instantiate(Unknown Source)
	at io.micronaut.context.DefaultBeanContext.resolveByBeanFactory(DefaultBeanContext.java:2309)
	at io.micronaut.context.DefaultBeanContext.doCreateBean(DefaultBeanContext.java:2279)
	at io.micronaut.context.DefaultBeanContext.doCreateBean(DefaultBeanContext.java:2291)
	at io.micronaut.context.DefaultBeanContext.createRegistration(DefaultBeanContext.java:3054)
	at io.micronaut.context.SingletonScope.getOrCreate(SingletonScope.java:80)
	at io.micronaut.context.DefaultBeanContext.findOrCreateSingletonBeanRegistration(DefaultBeanContext.java:2956)
	at io.micronaut.context.DefaultBeanContext.resolveBeanRegistration(DefaultBeanContext.java:2917)
	at io.micronaut.context.DefaultBeanContext.resolveBeanRegistration(DefaultBeanContext.java:2730)
	at io.micronaut.context.DefaultBeanContext.getBean(DefaultBeanContext.java:1729)
	at io.micronaut.context.DefaultBeanContext.getBean(DefaultBeanContext.java:856)
	at io.micronaut.context.DefaultBeanContext.getBean(DefaultBeanContext.java:848)
	at io.kestra.core.runners.StandAloneRunner.run(StandAloneRunner.java:38)
```
We should skip and warm about invalid flows

Issue comments:
This looks like a dependency issue as we no longer use reactivex.

>  io/reactivex/functions/BiConsumer
Yes, the goal of these issue is to kept the scheduler running even if a plugin could not be loaded, in that case, we just should log an error and don't load the flow behind. 

Pull request review comments:

Patch that resolved the issue:
diff --git a/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcFlowRepository.java b/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcFlowRepository.java
index 433fcebccf..77060e4ef7 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcFlowRepository.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcFlowRepository.java
@@ -23,8 +23,8 @@
 import io.micronaut.context.event.ApplicationEventPublisher;
 import io.micronaut.data.model.Pageable;
 import io.micronaut.inject.qualifiers.Qualifiers;
-import jakarta.inject.Singleton;
 import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
 import org.jooq.Record;
 import org.jooq.*;
 import org.jooq.impl.DSL;
@@ -33,7 +33,7 @@
 import jakarta.validation.ConstraintViolationException;
 import java.util.*;
 
-@Singleton
+@Slf4j
 public abstract class AbstractJdbcFlowRepository extends AbstractJdbcRepository implements FlowRepositoryInterface {
     private final QueueInterface<Flow> flowQueue;
     private final QueueInterface<Trigger> triggerQueue;
@@ -200,7 +200,20 @@ public List<Flow> findAllForAllTenants() {
                     .from(fromLastRevision(true))
                     .where(this.defaultFilter());
 
-                return this.jdbcRepository.fetch(select);
+                // findAllForAllTenants() is used in the backend, so we want it to work even if messy plugins exist.
+                // That's why we will try to deserialize each flow and log an error but not crash in case of exception.
+                List<Flow> flows = new ArrayList<>();
+                select.fetch().forEach(
+                    item -> {
+                        try {
+                            Flow flow = this.jdbcRepository.map(item);
+                            flows.add(flow);
+                        } catch (Exception e) {
+                            log.error("Unable to load the following flow:\n{}", item.get("value", String.class), e);
+                        }
+                    }
+                );
+                return flows;
             });
     }
 


Test patch: