From 1d4de46cf663c1dc9fa77337c9372977e5d6d0ad Mon Sep 17 00:00:00 2001
From: Ludovic DEHON <tchiot.ludo@gmail.com>
Date: Wed, 9 Feb 2022 21:10:13 +0100
Subject: [PATCH] feat(kafka-runner): used an in memory storage for flow &
 template to avoid performance issue on deserialization from rockdb

---
 .../io/kestra/core/models/flows/Flow.java     |  7 ++
 .../core/runners/FlowExecutorInterface.java   | 16 +++-
 .../core/runners/MemoryFlowExecutor.java      | 11 ++-
 .../java/io/kestra/core/tasks/flows/Flow.java |  3 +-
 .../io/kestra/core/tasks/flows/Template.java  | 19 ++---
 .../runner/kafka/KafkaFlowExecutor.java       | 70 ++++++++++++-----
 .../kestra/runner/kafka/KafkaScheduler.java   | 11 +--
 .../runner/kafka/KafkaTemplateExecutor.java   | 43 +++++++---
 .../kafka/executors/ExecutorFlowTrigger.java  |  7 +-
 .../runner/kafka/executors/ExecutorMain.java  | 27 +------
 .../runner/kafka/executors/ExecutorStore.java | 69 ++++++++++++++++
 .../executors/ExecutorTriggerCleaner.java     |  2 -
 .../services/KafkaStreamSourceService.java    | 35 +++------
 .../DeduplicationPurgeTransformer.java        |  3 +-
 .../streams/DeduplicationTransformer.java     |  3 +-
 .../ExecutorFromExecutionTransformer.java     |  3 +-
 .../streams/ExecutorJoinerTransformer.java    |  3 +-
 .../ExecutorKilledJoinerTransformer.java      |  3 +-
 .../kafka/streams/FlowJoinerTransformer.java  | 37 ++++-----
 .../FlowTriggerWithExecutionTransformer.java  | 50 ++++++------
 .../streams/FlowWithTriggerTransformer.java   | 24 ++----
 .../streams/GlobalInMemoryStateProcessor.java | 53 +++++++++++++
 .../streams/GlobalStateLockProcessor.java     | 17 ++--
 .../kafka/streams/GlobalStateProcessor.java   | 18 +++--
 .../kafka/streams/StateStoreTransformer.java  |  3 +-
 .../streams/WorkerInstanceTransformer.java    |  8 +-
 .../WorkerTaskExecutionTransformer.java       | 18 ++---
 .../runner/kafka/AbstractKafkaRunnerTest.java | 11 ++-
 .../runner/kafka/KafkaExecutorTest.java       | 78 ++++++++++++-------
 .../runner/kafka/KafkaFlowListenersTest.java  |  2 +
 .../kestra/runner/kafka/KafkaRunnerTest.java  |  6 +-
 .../runner/kafka/KafkaSchedulerTest.java      |  2 +
 .../kafka/services/SafeKeyValueStoreTest.java | 16 +++-
 .../kestra/runner/memory/MemoryExecutor.java  |  2 +-
 34 files changed, 431 insertions(+), 249 deletions(-)
 create mode 100644 runner-kafka/src/main/java/io/kestra/runner/kafka/executors/ExecutorStore.java
 create mode 100644 runner-kafka/src/main/java/io/kestra/runner/kafka/streams/GlobalInMemoryStateProcessor.java

diff --git a/core/src/main/java/io/kestra/core/models/flows/Flow.java b/core/src/main/java/io/kestra/core/models/flows/Flow.java
index 70be01e38f9..7f60b48522d 100644
--- a/core/src/main/java/io/kestra/core/models/flows/Flow.java
+++ b/core/src/main/java/io/kestra/core/models/flows/Flow.java
@@ -118,6 +118,13 @@ public static String uid(String namespace, String id, Optional<Integer> revision
         ));
     }
 
+    public static String uidWithoutRevision(String namespace, String id) {
+        return String.join("_", Arrays.asList(
+            namespace,
+            id
+        ));
+    }
+
     public static String uidWithoutRevision(Execution execution) {
         return String.join("_", Arrays.asList(
             execution.getNamespace(),
diff --git a/core/src/main/java/io/kestra/core/runners/FlowExecutorInterface.java b/core/src/main/java/io/kestra/core/runners/FlowExecutorInterface.java
index 00fed44e47d..64e8a347500 100644
--- a/core/src/main/java/io/kestra/core/runners/FlowExecutorInterface.java
+++ b/core/src/main/java/io/kestra/core/runners/FlowExecutorInterface.java
@@ -1,9 +1,23 @@
 package io.kestra.core.runners;
 
+import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.Flow;
 
+import java.util.Collection;
 import java.util.Optional;
 
 public interface FlowExecutorInterface {
-    Flow findById(String namespace, String id, Optional<Integer> revision, String fromNamespace, String fromId);
+    Collection<Flow> allLastVersion();
+
+    Optional<Flow> findById(String namespace, String id, Optional<Integer> revision, String fromNamespace, String fromId);
+
+    default Optional<Flow> findByExecution(Execution execution) {
+        return this.findById(
+            execution.getNamespace(),
+            execution.getFlowId(),
+            Optional.of(execution.getFlowRevision()),
+            null,
+            null
+        );
+    }
 }
diff --git a/core/src/main/java/io/kestra/core/runners/MemoryFlowExecutor.java b/core/src/main/java/io/kestra/core/runners/MemoryFlowExecutor.java
index 595c88c8076..83cf48b9447 100644
--- a/core/src/main/java/io/kestra/core/runners/MemoryFlowExecutor.java
+++ b/core/src/main/java/io/kestra/core/runners/MemoryFlowExecutor.java
@@ -3,6 +3,7 @@
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.repositories.FlowRepositoryInterface;
 
+import java.util.Collection;
 import java.util.Optional;
 
 public class MemoryFlowExecutor implements FlowExecutorInterface {
@@ -13,8 +14,12 @@ public MemoryFlowExecutor(FlowRepositoryInterface flowRepositoryInterface) {
     }
 
     @Override
-    public Flow findById(String namespace, String id, Optional<Integer> revision, String fromNamespace, String flowId) {
-        return flowRepositoryInterface.findById(namespace, id, revision)
-            .orElseThrow(() -> new IllegalStateException("Unable to find flow '" + namespace + "." + id + "' with revision + '" + revision + "'"));
+    public Collection<Flow> allLastVersion() {
+        return flowRepositoryInterface.findAll();
+    }
+
+    @Override
+    public Optional<Flow> findById(String namespace, String id, Optional<Integer> revision, String fromNamespace, String fromId) {
+        return flowRepositoryInterface.findById(namespace, id, revision);
     }
 }
diff --git a/core/src/main/java/io/kestra/core/tasks/flows/Flow.java b/core/src/main/java/io/kestra/core/tasks/flows/Flow.java
index feb1a1067e4..a516385f17a 100644
--- a/core/src/main/java/io/kestra/core/tasks/flows/Flow.java
+++ b/core/src/main/java/io/kestra/core/tasks/flows/Flow.java
@@ -122,7 +122,8 @@ public Execution createExecution(RunContext runContext, FlowExecutorInterface fl
             this.revision != null ? Optional.of(this.revision) : Optional.empty(),
             flowVars.get("namespace"),
             flowVars.get("id")
-        );
+        )
+            .orElseThrow(() -> new IllegalStateException("Unable to find flow '" + namespace + "." + id + "' with revision + '" + revision + "'"));
 
         return runnerUtils
             .newExecution(
diff --git a/core/src/main/java/io/kestra/core/tasks/flows/Template.java b/core/src/main/java/io/kestra/core/tasks/flows/Template.java
index 2267f18a900..ca8844ed97e 100644
--- a/core/src/main/java/io/kestra/core/tasks/flows/Template.java
+++ b/core/src/main/java/io/kestra/core/tasks/flows/Template.java
@@ -27,10 +27,7 @@
 import lombok.experimental.SuperBuilder;
 import lombok.extern.slf4j.Slf4j;
 
-import java.util.AbstractMap;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiFunction;
 import java.util.stream.Collectors;
@@ -190,12 +187,8 @@ public Template.Output outputs(RunContext runContext, Execution execution, TaskR
     protected io.kestra.core.models.templates.Template findTemplate(ApplicationContext applicationContext) throws IllegalVariableEvaluationException {
         TemplateExecutorInterface templateExecutor = applicationContext.getBean(TemplateExecutorInterface.class);
 
-        io.kestra.core.models.templates.Template template = templateExecutor.findById(this.namespace, this.templateId);
-        if (template == null) {
-            throw new IllegalVariableEvaluationException("Can't find flow template '" + this.namespace + "." + this.templateId + "'");
-        }
-
-        return template;
+        return templateExecutor.findById(this.namespace, this.templateId)
+            .orElseThrow(() -> new IllegalVariableEvaluationException("Can't find flow template '" + this.namespace + "." + this.templateId + "'"));
     }
 
     @SuperBuilder(toBuilder = true)
@@ -280,15 +273,15 @@ void onStartup(final StartupEvent event) {
     }
 
     public interface TemplateExecutorInterface {
-        io.kestra.core.models.templates.Template findById(String namespace, String templateId);
+        Optional<io.kestra.core.models.templates.Template> findById(String namespace, String templateId);
     }
 
     public static class MemoryTemplateExecutor implements io.kestra.core.tasks.flows.Template.TemplateExecutorInterface {
         @Inject
         private TemplateRepositoryInterface templateRepository;
 
-        public io.kestra.core.models.templates.Template findById(String namespace, String templateId) {
-            return this.templateRepository.findById(namespace, templateId).orElse(null);
+        public Optional<io.kestra.core.models.templates.Template> findById(String namespace, String templateId) {
+            return this.templateRepository.findById(namespace, templateId);
         }
     }
 
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
index 1c6d59bc573..99349df2a47 100644
--- a/runner-kafka/src/main/java/io/kestra/runner/kafka/KafkaFlowExecutor.java
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/KafkaFlowExecutor.java
@@ -3,34 +3,68 @@
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.runners.FlowExecutorInterface;
 import io.kestra.core.services.FlowService;
-import io.kestra.runner.kafka.services.SafeKeyValueStore;
-import io.micronaut.context.ApplicationContext;
+import io.kestra.core.utils.Await;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
 
-import java.util.Optional;
+import java.time.Duration;
+import java.util.*;
+import java.util.stream.Collectors;
 
 @Slf4j
+@Singleton
 public class KafkaFlowExecutor implements FlowExecutorInterface {
-    private final FlowService flowService;
-    private final SafeKeyValueStore<String, Flow> store;
+    @Inject
+    private FlowService flowService;
+    private Map<String, Flow> flows;
+    private Map<String, Flow> flowsLast;
 
-    public KafkaFlowExecutor(ReadOnlyKeyValueStore<String, Flow> store, String name, ApplicationContext applicationContext) {
-        this.store = new SafeKeyValueStore<>(store, name);
-        this.flowService = applicationContext.getBean(FlowService.class);
+    public synchronized void setFlows(List<Flow> flows) {
+        this.flows = flows
+            .stream()
+            .map(flow -> new AbstractMap.SimpleEntry<>(
+                flow.uid(),
+                flow
+            ))
+            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+        this.flowsLast = flowService.keepLastVersion(flows)
+            .stream()
+            .map(flow -> new AbstractMap.SimpleEntry<>(
+                flow.uidWithoutRevision(),
+                flow
+            ))
+            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+    }
+
+    @SneakyThrows
+    private void await() {
+        if (flows == null) {
+            Await.until(() -> this.flows != null, Duration.ofMillis(100), Duration.ofMinutes(5));
+        }
+    }
+
+    @Override
+    public Collection<Flow> allLastVersion() {
+        this.await();
+
+        return this.flowsLast.values();
     }
 
     @Override
-    public Flow findById(String namespace, String id, Optional<Integer> revision, String fromNamespace, String fromId) {
+    public Optional<Flow> findById(String namespace, String id, Optional<Integer> revision, String fromNamespace, String fromId) {
+        this.await();
+
         if (revision.isPresent()) {
-            return this.store.get(Flow.uid(namespace, id, revision))
-                .orElseThrow(() -> new IllegalStateException("Unable to find flow '" + namespace + "." + id + "'"));
-        } else {
-            return flowService.keepLastVersion(
-                this.store.all(),
-                namespace,
-                id
-            );
+            String uid = Flow.uid(namespace, id, revision);
+
+            return Optional.ofNullable(this.flows.get(uid));
         }
+
+        String uid = Flow.uidWithoutRevision(namespace, id);
+
+        return Optional.ofNullable(this.flowsLast.get(uid));
     }
 }
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/KafkaScheduler.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/KafkaScheduler.java
index 8478729bfb0..8f898891a77 100644
--- a/runner-kafka/src/main/java/io/kestra/runner/kafka/KafkaScheduler.java
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/KafkaScheduler.java
@@ -11,17 +11,20 @@
 import io.kestra.core.schedulers.AbstractScheduler;
 import io.kestra.core.schedulers.DefaultScheduler;
 import io.kestra.core.schedulers.SchedulerExecutionWithTrigger;
-import io.kestra.core.services.ConditionService;
 import io.kestra.core.services.FlowListenersInterface;
 import io.kestra.core.utils.IdUtils;
 import io.kestra.runner.kafka.configs.TopicsConfig;
 import io.kestra.runner.kafka.serializers.JsonSerde;
-import io.kestra.runner.kafka.services.*;
+import io.kestra.runner.kafka.services.KafkaAdminService;
+import io.kestra.runner.kafka.services.KafkaProducerService;
+import io.kestra.runner.kafka.services.KafkaStreamService;
+import io.kestra.runner.kafka.services.KafkaStreamsBuilder;
 import io.kestra.runner.kafka.streams.GlobalStateLockProcessor;
 import io.kestra.runner.kafka.streams.GlobalStateProcessor;
 import io.micronaut.context.ApplicationContext;
 import io.micronaut.context.annotation.Replaces;
 import io.micronaut.inject.qualifiers.Qualifiers;
+import jakarta.inject.Singleton;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -30,15 +33,13 @@
 import org.apache.kafka.streams.StoreQueryParameters;
 import org.apache.kafka.streams.StreamsBuilder;
 import org.apache.kafka.streams.Topology;
-import org.apache.kafka.streams.kstream.*;
-import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
+import org.apache.kafka.streams.kstream.Consumed;
 import org.apache.kafka.streams.state.QueryableStoreTypes;
 import org.apache.kafka.streams.state.Stores;
 
 import java.time.Duration;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
-import jakarta.inject.Singleton;
 
 @KafkaQueueEnabled
 @Singleton
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/KafkaTemplateExecutor.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/KafkaTemplateExecutor.java
index 1ed8261290c..2e792bfd699 100644
--- a/runner-kafka/src/main/java/io/kestra/runner/kafka/KafkaTemplateExecutor.java
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/KafkaTemplateExecutor.java
@@ -1,24 +1,49 @@
 package io.kestra.runner.kafka;
 
-import io.kestra.runner.kafka.services.SafeKeyValueStore;
+import io.kestra.core.models.templates.Template;
+import io.kestra.core.utils.Await;
 import io.micronaut.context.annotation.Replaces;
 import io.micronaut.context.annotation.Requires;
+import jakarta.inject.Singleton;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
-import io.kestra.core.models.templates.Template;
+
+import java.time.Duration;
+import java.util.AbstractMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
 
 @Slf4j
 @KafkaQueueEnabled
 @Replaces(io.kestra.core.tasks.flows.Template.MemoryTemplateExecutor.class)
-@Requires(property = "kestra.server-type", value = "EXECUTOR")
+@Requires(property = "kestra.server-type", pattern = "(EXECUTOR|STANDALONE)")
+@Singleton
 public class KafkaTemplateExecutor implements io.kestra.core.tasks.flows.Template.TemplateExecutorInterface {
-    private final SafeKeyValueStore<String, Template> store;
+    private Map<String, Template> templates;
+
 
-    public KafkaTemplateExecutor(ReadOnlyKeyValueStore<String, Template> store, String name) {
-        this.store = new SafeKeyValueStore<>(store, name);
+    public synchronized void setTemplates(List<Template> templates) {
+        this.templates = templates
+            .stream()
+            .map(template -> new AbstractMap.SimpleEntry<>(
+                template.uid(),
+                template
+            ))
+            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
     }
 
-    public Template findById(String namespace, String templateId) {
-        return this.store.get(Template.uid(namespace, templateId)).orElse(null);
+    @SneakyThrows
+    private void await() {
+        if (templates == null) {
+            Await.until(() -> this.templates != null, Duration.ofMillis(100), Duration.ofMinutes(5));
+        }
+    }
+
+    public Optional<Template> findById(String namespace, String templateId) {
+        this.await();
+
+        return Optional.ofNullable(this.templates.get(Template.uid(namespace, templateId)));
     }
 }
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/executors/ExecutorFlowTrigger.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/executors/ExecutorFlowTrigger.java
index b9b6e38e534..425fc85a6f8 100644
--- a/runner-kafka/src/main/java/io/kestra/runner/kafka/executors/ExecutorFlowTrigger.java
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/executors/ExecutorFlowTrigger.java
@@ -3,10 +3,10 @@
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.triggers.multipleflows.MultipleConditionWindow;
 import io.kestra.core.services.FlowService;
+import io.kestra.runner.kafka.KafkaFlowExecutor;
 import io.kestra.runner.kafka.KafkaQueueEnabled;
 import io.kestra.runner.kafka.serializers.JsonSerde;
 import io.kestra.runner.kafka.services.KafkaAdminService;
-import io.kestra.runner.kafka.services.KafkaStreamSourceService;
 import io.kestra.runner.kafka.services.KafkaStreamsBuilder;
 import io.kestra.runner.kafka.streams.FlowTriggerWithExecutionTransformer;
 import jakarta.inject.Inject;
@@ -34,13 +34,11 @@ public class ExecutorFlowTrigger implements KafkaExecutorInterface {
     private FlowService flowService;
 
     @Inject
-    private KafkaStreamSourceService kafkaStreamSourceService;
+    private KafkaFlowExecutor kafkaFlowExecutor;
 
     public StreamsBuilder topology() {
         StreamsBuilder builder = new KafkaStreamsBuilder();
 
-        kafkaStreamSourceService.flowGlobalKTable(builder);
-
         // trigger
         builder.addStateStore(
             Stores.keyValueStoreBuilder(
@@ -62,6 +60,7 @@ public StreamsBuilder topology() {
             .transformValues(
                 () -> new FlowTriggerWithExecutionTransformer(
                     TRIGGER_MULTIPLE_STATE_STORE_NAME,
+                    kafkaFlowExecutor,
                     flowService
                 ),
                 Named.as("ExecutorFlowTrigger.transformToExecutionList"),
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/executors/ExecutorMain.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/executors/ExecutorMain.java
index a1e21a0bf62..6b5108f1a19 100644
--- a/runner-kafka/src/main/java/io/kestra/runner/kafka/executors/ExecutorMain.java
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/executors/ExecutorMain.java
@@ -51,8 +51,6 @@ public class ExecutorMain implements KafkaExecutorInterface {
     private static final String TRIGGER_DEDUPLICATION_STATE_STORE_NAME = "trigger_deduplication";
     private static final String NEXTS_DEDUPLICATION_STATE_STORE_NAME = "next_deduplication";
 
-    protected FlowExecutorInterface flowExecutorInterface;
-
     @Inject
     private KafkaAdminService kafkaAdminService;
 
@@ -77,23 +75,8 @@ public class ExecutorMain implements KafkaExecutorInterface {
     @Inject
     private RunContextFactory runContextFactory;
 
-    @Override
-    public void onCreated(ApplicationContext applicationContext, KafkaStreamService.Stream stream) {
-        applicationContext.registerSingleton(new KafkaTemplateExecutor(
-            stream.store(StoreQueryParameters.fromNameAndType("template", QueryableStoreTypes.keyValueStore())),
-            "template"
-        ));
-
-        if (flowExecutorInterface == null) {
-            this.flowExecutorInterface = new KafkaFlowExecutor(
-                stream.store(StoreQueryParameters.fromNameAndType("flow", QueryableStoreTypes.keyValueStore())),
-                "flow",
-                applicationContext
-            );
-        }
-
-        applicationContext.registerSingleton(this.flowExecutorInterface);
-    }
+    @Inject
+    private KafkaFlowExecutor kafkaFlowExecutor;
 
     public StreamsBuilder topology() {
         StreamsBuilder builder = new KafkaStreamsBuilder();
@@ -138,8 +121,6 @@ public StreamsBuilder topology() {
         KStream<String, Executor> executionKStream = this.joinWorkerResult(workerTaskResultKStream, executionWithKilled);
 
         // handle state on execution
-        GlobalKTable<String, Flow> flowKTable = kafkaStreamSourceService.flowGlobalKTable(builder);
-        GlobalKTable<String, Template> templateKTable = kafkaStreamSourceService.templateGlobalKTable(builder);
         KStream<String, Executor> stream = kafkaStreamSourceService.executorWithFlow(executionKStream, true);
 
         stream = this.handleExecutor(stream);
@@ -380,7 +361,7 @@ private void toExecutorFlowTriggerTopic(KStream<String, Executor> stream) {
             )
             .filter((key, value) -> value != null, Named.as("HandleExecutorFlowTriggerTopic.deduplicationNotNull"))
             .flatTransform(
-                () -> new FlowWithTriggerTransformer(flowService),
+                () -> new FlowWithTriggerTransformer(kafkaFlowExecutor, flowService),
                 Named.as("HandleExecutorFlowTriggerTopic.flatMapToExecutorFlowTrigger")
             )
             .to(
@@ -563,7 +544,7 @@ private void handleWorkerTaskExecution(KTable<String, WorkerTaskExecution> worke
                 Named.as("HandleWorkerTaskExecution.isTerminated")
             )
             .transformValues(
-                () -> new WorkerTaskExecutionTransformer(runContextFactory, workerTaskExecutionKTable.queryableStoreName()),
+                () -> new WorkerTaskExecutionTransformer(runContextFactory, workerTaskExecutionKTable.queryableStoreName(), kafkaFlowExecutor),
                  Named.as("HandleWorkerTaskExecution.transform"),
                 workerTaskExecutionKTable.queryableStoreName()
             )
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/executors/ExecutorStore.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/executors/ExecutorStore.java
new file mode 100644
index 00000000000..e349a126e55
--- /dev/null
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/executors/ExecutorStore.java
@@ -0,0 +1,69 @@
+package io.kestra.runner.kafka.executors;
+
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.templates.Template;
+import io.kestra.runner.kafka.KafkaFlowExecutor;
+import io.kestra.runner.kafka.KafkaQueueEnabled;
+import io.kestra.runner.kafka.KafkaTemplateExecutor;
+import io.kestra.runner.kafka.serializers.JsonSerde;
+import io.kestra.runner.kafka.services.KafkaAdminService;
+import io.kestra.runner.kafka.services.KafkaStreamsBuilder;
+import io.kestra.runner.kafka.streams.GlobalInMemoryStateProcessor;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.kstream.Consumed;
+import org.apache.kafka.streams.state.Stores;
+
+@KafkaQueueEnabled
+@Singleton
+@Slf4j
+public class ExecutorStore implements KafkaExecutorInterface {
+    public static final String FLOW_STATE_STORE_NAME = "flow";
+    public static final String TEMPLATE_STATE_STORE_NAME = "template";
+
+    @Inject
+    private KafkaAdminService kafkaAdminService;
+
+    @Inject
+    private KafkaFlowExecutor kafkaFlowExecutor;
+
+    @Inject
+    private KafkaTemplateExecutor kafkaTemplateExecutor;
+
+    public StreamsBuilder topology() {
+        StreamsBuilder builder = new KafkaStreamsBuilder();
+
+        builder.addGlobalStore(
+            Stores.keyValueStoreBuilder(
+                Stores.persistentKeyValueStore(FLOW_STATE_STORE_NAME),
+                Serdes.String(),
+                JsonSerde.of(Flow.class)
+            ),
+            kafkaAdminService.getTopicName(Flow.class),
+            Consumed.with(Serdes.String(), JsonSerde.of(Flow.class)).withName("GlobalStore.Flow"),
+            () -> new GlobalInMemoryStateProcessor<>(
+                FLOW_STATE_STORE_NAME,
+                flows -> kafkaFlowExecutor.setFlows(flows)
+            )
+        );
+
+        builder.addGlobalStore(
+            Stores.keyValueStoreBuilder(
+                Stores.persistentKeyValueStore(TEMPLATE_STATE_STORE_NAME),
+                Serdes.String(),
+                JsonSerde.of(Template.class)
+            ),
+            kafkaAdminService.getTopicName(Template.class),
+            Consumed.with(Serdes.String(), JsonSerde.of(Template.class)).withName("GlobalStore.Template"),
+            () -> new GlobalInMemoryStateProcessor<>(
+                TEMPLATE_STATE_STORE_NAME,
+                templates -> kafkaTemplateExecutor.setTemplates(templates)
+            )
+        );
+
+        return builder;
+    }
+}
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/executors/ExecutorTriggerCleaner.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/executors/ExecutorTriggerCleaner.java
index 8e02ee0467f..0b891034c64 100644
--- a/runner-kafka/src/main/java/io/kestra/runner/kafka/executors/ExecutorTriggerCleaner.java
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/executors/ExecutorTriggerCleaner.java
@@ -40,8 +40,6 @@ public StreamsBuilder topology() {
 
         KStream<String, Executor> executorKStream = kafkaStreamSourceService.executorKStream(builder);
 
-        kafkaStreamSourceService.flowGlobalKTable(builder);
-        kafkaStreamSourceService.templateGlobalKTable(builder);
         KStream<String, Executor> executionWithFlowKStream = kafkaStreamSourceService.executorWithFlow(executorKStream, false);
 
         GlobalKTable<String, Trigger> triggerGlobalKTable = kafkaStreamSourceService.triggerGlobalKTable(builder);
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
index 0a8eee91ca5..019a4ac9fcc 100644
--- 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
@@ -1,12 +1,14 @@
 package io.kestra.runner.kafka.services;
 
-import io.kestra.core.models.flows.Flow;
-import io.kestra.core.models.templates.Template;
 import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.runners.Executor;
+import io.kestra.core.runners.FlowExecutorInterface;
 import io.kestra.core.services.TaskDefaultService;
+import io.kestra.core.tasks.flows.Template;
 import io.kestra.runner.kafka.serializers.JsonSerde;
 import io.kestra.runner.kafka.streams.FlowJoinerTransformer;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.common.utils.Bytes;
 import org.apache.kafka.streams.StreamsBuilder;
@@ -14,9 +16,6 @@
 import org.apache.kafka.streams.state.KeyValueStore;
 import org.slf4j.Logger;
 
-import jakarta.inject.Inject;
-import jakarta.inject.Singleton;
-
 @Singleton
 public class KafkaStreamSourceService {
     public static final String TOPIC_FLOWLAST = "flowlast";
@@ -28,27 +27,11 @@ public class KafkaStreamSourceService {
     @Inject
     private TaskDefaultService taskDefaultService;
 
-    public GlobalKTable<String, Flow> flowGlobalKTable(StreamsBuilder builder) {
-        return builder
-            .globalTable(
-                kafkaAdminService.getTopicName(Flow.class),
-                Consumed.with(Serdes.String(), JsonSerde.of(Flow.class)).withName("GlobalKTable.Flow"),
-                Materialized.<String, Flow, KeyValueStore<Bytes, byte[]>>as("flow")
-                    .withKeySerde(Serdes.String())
-                    .withValueSerde(JsonSerde.of(Flow.class))
-            );
-    }
+    @Inject
+    private FlowExecutorInterface flowExecutorInterface;
 
-    public GlobalKTable<String, Template> templateGlobalKTable(StreamsBuilder builder) {
-        return builder
-            .globalTable(
-                kafkaAdminService.getTopicName(Template.class),
-                Consumed.with(Serdes.String(), JsonSerde.of(Template.class)).withName("GlobalKTable.Template"),
-                Materialized.<String, Template, KeyValueStore<Bytes, byte[]>>as("template")
-                    .withKeySerde(Serdes.String())
-                    .withValueSerde(JsonSerde.of(Template.class))
-            );
-    }
+    @Inject
+    private Template.TemplateExecutorInterface templateExecutorInterface;
 
     public KStream<String, Executor> executorKStream(StreamsBuilder builder) {
         return builder
@@ -73,7 +56,7 @@ public KStream<String, Executor> executorWithFlow(KStream<String, Executor> exec
         return executionKStream
             .filter((key, value) -> value != null, Named.as("ExecutorWithFlow.filterNotNull"))
             .transformValues(
-                () -> new FlowJoinerTransformer(withDefaults, taskDefaultService)
+                () -> new FlowJoinerTransformer(flowExecutorInterface, templateExecutorInterface, withDefaults, taskDefaultService)
             );
     }
 
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/DeduplicationPurgeTransformer.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/DeduplicationPurgeTransformer.java
index b45e5809165..147ccffa3ae 100644
--- a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/DeduplicationPurgeTransformer.java
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/DeduplicationPurgeTransformer.java
@@ -21,9 +21,8 @@ public DeduplicationPurgeTransformer(
     }
 
     @Override
-    @SuppressWarnings("unchecked")
     public void init(final ProcessorContext context) {
-        this.store = (KeyValueStore<String, SV>) context.getStateStore(this.storeName);
+        this.store = context.getStateStore(this.storeName);
     }
 
     @Override
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/DeduplicationTransformer.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/DeduplicationTransformer.java
index 7d3b2ff1707..da59273a74e 100644
--- a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/DeduplicationTransformer.java
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/DeduplicationTransformer.java
@@ -27,9 +27,8 @@ public DeduplicationTransformer(
     }
 
     @Override
-    @SuppressWarnings("unchecked")
     public void init(final ProcessorContext context) {
-        this.store = (KeyValueStore<String, SV>) context.getStateStore(this.storeName);
+        this.store = context.getStateStore(this.storeName);
     }
 
     @Override
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/ExecutorFromExecutionTransformer.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/ExecutorFromExecutionTransformer.java
index f4d5145c983..91fe7f4b4d6 100644
--- a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/ExecutorFromExecutionTransformer.java
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/ExecutorFromExecutionTransformer.java
@@ -19,10 +19,9 @@ public ExecutorFromExecutionTransformer(String storeName) {
     }
 
     @Override
-    @SuppressWarnings("unchecked")
     public void init(final ProcessorContext context) {
         this.context = context;
-        this.store = (KeyValueStore<String, Executor>) context.getStateStore(this.storeName);
+        this.store = context.getStateStore(this.storeName);
     }
 
     @Override
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/ExecutorJoinerTransformer.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/ExecutorJoinerTransformer.java
index f0f4987b491..ba5dbd1b9c5 100644
--- a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/ExecutorJoinerTransformer.java
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/ExecutorJoinerTransformer.java
@@ -22,10 +22,9 @@ public ExecutorJoinerTransformer(String storeName, MetricRegistry metricRegistry
     }
 
     @Override
-    @SuppressWarnings("unchecked")
     public void init(final ProcessorContext context) {
         this.context = context;
-        this.store = (KeyValueStore<String, Executor>) context.getStateStore(this.storeName);
+        this.store = context.getStateStore(this.storeName);
     }
 
     @Override
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/ExecutorKilledJoinerTransformer.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/ExecutorKilledJoinerTransformer.java
index 34660ce0f9a..73ff99c3f77 100644
--- a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/ExecutorKilledJoinerTransformer.java
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/ExecutorKilledJoinerTransformer.java
@@ -19,9 +19,8 @@ public ExecutorKilledJoinerTransformer(String storeName) {
     }
 
     @Override
-    @SuppressWarnings("unchecked")
     public void init(final ProcessorContext context) {
-        this.store = (KeyValueStore<String, Executor>) context.getStateStore(this.storeName);
+        this.store = context.getStateStore(this.storeName);
     }
 
     @Override
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/FlowJoinerTransformer.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/FlowJoinerTransformer.java
index 16ffa19f085..74d659a7388 100644
--- a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/FlowJoinerTransformer.java
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/FlowJoinerTransformer.java
@@ -3,14 +3,12 @@
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.runners.Executor;
+import io.kestra.core.runners.FlowExecutorInterface;
 import io.kestra.core.services.TaskDefaultService;
 import io.kestra.core.tasks.flows.Template;
-import io.kestra.runner.kafka.services.SafeKeyValueStore;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.streams.kstream.ValueTransformerWithKey;
 import org.apache.kafka.streams.processor.ProcessorContext;
-import org.apache.kafka.streams.state.KeyValueStore;
-import org.apache.kafka.streams.state.ValueAndTimestamp;
 
 import java.util.Optional;
 
@@ -18,48 +16,41 @@
 public class FlowJoinerTransformer implements ValueTransformerWithKey<String, Executor, Executor> {
     private final boolean withDefaults;
     private final TaskDefaultService taskDefaultService;
+    private final FlowExecutorInterface flowExecutorInterface;
+    private final Template.TemplateExecutorInterface templateExecutorInterface;
 
-    private SafeKeyValueStore<String, ValueAndTimestamp<Flow>> flowStore;
-    private SafeKeyValueStore<String, ValueAndTimestamp<io.kestra.core.models.templates.Template>> templateStore;
-
-    public FlowJoinerTransformer(boolean withDefaults, TaskDefaultService taskDefaultService) {
+    public FlowJoinerTransformer(FlowExecutorInterface flowExecutorInterface, Template.TemplateExecutorInterface templateExecutorInterface, boolean withDefaults, TaskDefaultService taskDefaultService) {
+        this.flowExecutorInterface = flowExecutorInterface;
+        this.templateExecutorInterface = templateExecutorInterface;
         this.withDefaults = withDefaults;
         this.taskDefaultService = taskDefaultService;
     }
 
     @Override
-    @SuppressWarnings("unchecked")
     public void init(final ProcessorContext context) {
-        var flowStore = (KeyValueStore<String, ValueAndTimestamp<Flow>>) context.getStateStore("flow");
-        this.flowStore = new SafeKeyValueStore<>(flowStore, flowStore.name());
 
-        var templateStore = (KeyValueStore<String, ValueAndTimestamp<io.kestra.core.models.templates.Template>>) context.getStateStore("template");
-        this.templateStore = new SafeKeyValueStore<>(templateStore, templateStore.name());
     }
 
     @Override
     public Executor transform(String key, Executor executor) {
-        Optional<ValueAndTimestamp<Flow>> flowState = this.flowStore.get(Flow.uid(executor.getExecution()));
+        Optional<Flow> flowState = flowExecutorInterface.findByExecution(executor.getExecution());
 
-        if (flowState.isEmpty() || flowState.get().value() == null) {
+        if (flowState.isEmpty()) {
             return null;
         }
 
-        Flow flow = flowState.get().value();
+        Flow flow = flowState.get();
 
         if (!withDefaults) {
             return executor.withFlow(flow);
         }
 
         try {
-            flow = Template.injectTemplate(flow, executor.getExecution(), (namespace, id) -> {
-                String templateUid = io.kestra.core.models.templates.Template.uid(
-                    namespace,
-                    id
-                );
-
-                return this.templateStore.get(templateUid).map(ValueAndTimestamp::value).orElse(null);
-            });
+            flow = Template.injectTemplate(
+                flow,
+                executor.getExecution(),
+                (namespace, id) -> this.templateExecutorInterface.findById(namespace, id).orElse(null)
+            );
         } catch (InternalException e) {
             log.warn("Failed to inject template",  e);
         }
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/FlowTriggerWithExecutionTransformer.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/FlowTriggerWithExecutionTransformer.java
index a8cf9d5daa5..9fbf970f818 100644
--- a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/FlowTriggerWithExecutionTransformer.java
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/FlowTriggerWithExecutionTransformer.java
@@ -1,17 +1,15 @@
 package io.kestra.runner.kafka.streams;
 
-import io.kestra.runner.kafka.services.SafeKeyValueStore;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.streams.kstream.ValueTransformer;
-import org.apache.kafka.streams.processor.ProcessorContext;
-import org.apache.kafka.streams.state.KeyValueStore;
-import org.apache.kafka.streams.state.ValueAndTimestamp;
 import io.kestra.core.models.executions.Execution;
-import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.triggers.multipleflows.MultipleConditionStorageInterface;
 import io.kestra.core.models.triggers.multipleflows.MultipleConditionWindow;
 import io.kestra.core.services.FlowService;
+import io.kestra.runner.kafka.KafkaFlowExecutor;
 import io.kestra.runner.kafka.KafkaMultipleConditionStorage;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.streams.kstream.ValueTransformer;
+import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.state.KeyValueStore;
 
 import java.util.List;
 import java.util.stream.Stream;
@@ -20,21 +18,19 @@
 public class FlowTriggerWithExecutionTransformer implements ValueTransformer<ExecutorFlowTrigger, List<Execution>> {
     private final String multipleStoreName;
     private final FlowService flowService;
+    private final KafkaFlowExecutor kafkaFlowExecutor;
 
     private KeyValueStore<String, MultipleConditionWindow> multipleStore;
-    private SafeKeyValueStore<String, ValueAndTimestamp<Flow>> flowStore;
 
-    public FlowTriggerWithExecutionTransformer(String multipleStoreName, FlowService flowService) {
+    public FlowTriggerWithExecutionTransformer(String multipleStoreName, KafkaFlowExecutor kafkaFlowExecutor, FlowService flowService) {
         this.multipleStoreName = multipleStoreName;
+        this.kafkaFlowExecutor = kafkaFlowExecutor;
         this.flowService = flowService;
     }
 
     @Override
-    @SuppressWarnings("unchecked")
     public void init(final ProcessorContext context) {
-        var flowStore = (KeyValueStore<String, ValueAndTimestamp<Flow>>) context.getStateStore("flow");
-        this.flowStore = new SafeKeyValueStore<>(flowStore, flowStore.name());
-        this.multipleStore = (KeyValueStore<String, MultipleConditionWindow>) context.getStateStore(this.multipleStoreName);
+        this.multipleStore = context.getStateStore(this.multipleStoreName);
     }
 
     @Override
@@ -42,21 +38,19 @@ public List<Execution> transform(final ExecutorFlowTrigger value) {
         MultipleConditionStorageInterface multipleConditionStorage = new KafkaMultipleConditionStorage(this.multipleStore);
 
         // multiple conditions storage
-        this.flowStore.get(Flow.uid(value.getExecution()))
-            .ifPresent(flowValueAndTimestamp -> {
-            Flow executionFlow = flowValueAndTimestamp.value();
-
-            flowService
-                .multipleFlowTrigger(
-                    Stream.of(value.getFlowHavingTrigger()),
-                    executionFlow,
-                    value.getExecution(),
-                    multipleConditionStorage
-                )
-                .forEach(triggerExecutionWindow -> {
-                    this.multipleStore.put(triggerExecutionWindow.uid(), triggerExecutionWindow);
-                });
-        });
+        kafkaFlowExecutor.findByExecution(value.getExecution())
+            .ifPresent(flow -> {
+                flowService
+                    .multipleFlowTrigger(
+                        Stream.of(value.getFlowHavingTrigger()),
+                        flow,
+                        value.getExecution(),
+                        multipleConditionStorage
+                    )
+                    .forEach(triggerExecutionWindow -> {
+                        this.multipleStore.put(triggerExecutionWindow.uid(), triggerExecutionWindow);
+                    });
+            });
 
         List<Execution> triggeredExecutions = flowService.flowTriggerExecution(
             Stream.of(value.getFlowHavingTrigger()),
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/FlowWithTriggerTransformer.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/FlowWithTriggerTransformer.java
index 66a98ea166e..679bdcf45d8 100644
--- a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/FlowWithTriggerTransformer.java
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/FlowWithTriggerTransformer.java
@@ -1,43 +1,33 @@
 package io.kestra.runner.kafka.streams;
 
-import io.kestra.core.models.flows.Flow;
 import io.kestra.core.runners.Executor;
 import io.kestra.core.services.FlowService;
-import io.kestra.runner.kafka.services.SafeKeyValueStore;
+import io.kestra.runner.kafka.KafkaFlowExecutor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.kstream.Transformer;
 import org.apache.kafka.streams.processor.ProcessorContext;
-import org.apache.kafka.streams.state.KeyValueStore;
-import org.apache.kafka.streams.state.ValueAndTimestamp;
 
-import java.util.List;
 import java.util.stream.Collectors;
 
 @Slf4j
 public class FlowWithTriggerTransformer implements Transformer<String, Executor, Iterable<KeyValue<String, ExecutorFlowTrigger>>> {
     private final FlowService flowService;
+    private final KafkaFlowExecutor kafkaFlowExecutor;
 
-    private SafeKeyValueStore<String, ValueAndTimestamp<Flow>> flowStore;
-
-    public FlowWithTriggerTransformer(FlowService flowService) {
+    public FlowWithTriggerTransformer(KafkaFlowExecutor kafkaFlowExecutor, FlowService flowService) {
         this.flowService = flowService;
+        this.kafkaFlowExecutor = kafkaFlowExecutor;
     }
 
     @Override
-    @SuppressWarnings("unchecked")
-    public void init(final ProcessorContext context) {
-        var store = (KeyValueStore<String, ValueAndTimestamp<Flow>>) context.getStateStore("flow");
-        this.flowStore = new SafeKeyValueStore<>(store, store.name());
+    public void init(ProcessorContext context) {
+
     }
 
     @Override
     public Iterable<KeyValue<String, ExecutorFlowTrigger>> transform(String key, Executor value) {
-        List<Flow> allFlows = flowService
-            .keepLastVersion(flowStore.all().map(ValueAndTimestamp::value))
-            .collect(Collectors.toList());
-
-        return flowService.flowWithFlowTrigger(allFlows.stream())
+        return flowService.flowWithFlowTrigger(kafkaFlowExecutor.allLastVersion().stream())
             .stream()
             .map(f -> KeyValue.pair(f.getFlow().uidWithoutRevision(), new ExecutorFlowTrigger(f.getFlow(), value.getExecution())))
             .collect(Collectors.toList());
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/GlobalInMemoryStateProcessor.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/GlobalInMemoryStateProcessor.java
new file mode 100644
index 00000000000..e6b520fae16
--- /dev/null
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/GlobalInMemoryStateProcessor.java
@@ -0,0 +1,53 @@
+package io.kestra.runner.kafka.streams;
+
+import io.kestra.runner.kafka.services.SafeKeyValueStore;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.streams.processor.api.Processor;
+import org.apache.kafka.streams.processor.api.ProcessorContext;
+import org.apache.kafka.streams.processor.api.Record;
+import org.apache.kafka.streams.state.KeyValueStore;
+
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+
+@Slf4j
+public class GlobalInMemoryStateProcessor<T> implements Processor<String, T, Void, Void> {
+    private final String storeName;
+    private final Consumer<List<T>> consumer;
+    private KeyValueStore<String, T> store;
+    private SafeKeyValueStore<String, T> safeStore;
+
+    public GlobalInMemoryStateProcessor(String storeName, Consumer<List<T>> consumer) {
+        this.storeName = storeName;
+        this.consumer = consumer;
+    }
+
+    @Override
+    public void init(ProcessorContext<Void, Void> context) {
+        this.store = context.getStateStore(this.storeName);
+        this.safeStore = new SafeKeyValueStore<>(this.store, this.store.name());
+
+        this.send();
+    }
+
+    @Override
+    public void process(Record<String, T> record) {
+        if (record.value() == null) {
+            this.store.delete(record.key());
+        } else {
+            this.store.put(record.key(), record.value());
+        }
+
+        this.send();
+    }
+
+    private void send() {
+        consumer.accept(this.safeStore.all().collect(Collectors.toList()));
+    }
+
+    @Override
+    public void close() {
+
+    }
+}
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/GlobalStateLockProcessor.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/GlobalStateLockProcessor.java
index 9b744ce0fd0..f20cd7b4ee9 100644
--- a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/GlobalStateLockProcessor.java
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/GlobalStateLockProcessor.java
@@ -1,14 +1,15 @@
 package io.kestra.runner.kafka.streams;
 
 import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.streams.processor.Processor;
-import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.processor.api.Processor;
+import org.apache.kafka.streams.processor.api.ProcessorContext;
+import org.apache.kafka.streams.processor.api.Record;
 import org.apache.kafka.streams.state.KeyValueStore;
 
 import java.util.Map;
 
 @Slf4j
-public class GlobalStateLockProcessor<T> implements Processor<String, T> {
+public class GlobalStateLockProcessor<T> implements Processor<String, T, Void, Void> {
     private final String storeName;
     private final Map<String, T> lock;
     private KeyValueStore<String, T> store;
@@ -25,14 +26,14 @@ public void init(ProcessorContext context) {
     }
 
     @Override
-    public void process(String key, T value) {
-        if (value == null) {
-            this.store.delete(key);
+    public void process(Record<String, T> record) {
+        if (record.value() == null) {
+            this.store.delete(record.key());
         } else {
-            this.store.put(key, value);
+            this.store.put(record.key(), record.value());
         }
 
-        this.lock.remove(key);
+        this.lock.remove(record.key());
     }
 
     @Override
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/GlobalStateProcessor.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/GlobalStateProcessor.java
index 097547c69d3..0a68d34d892 100644
--- a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/GlobalStateProcessor.java
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/GlobalStateProcessor.java
@@ -2,8 +2,9 @@
 
 import com.google.common.collect.Streams;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.streams.processor.Processor;
-import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.processor.api.Processor;
+import org.apache.kafka.streams.processor.api.ProcessorContext;
+import org.apache.kafka.streams.processor.api.Record;
 import org.apache.kafka.streams.state.KeyValueIterator;
 import org.apache.kafka.streams.state.KeyValueStore;
 
@@ -12,7 +13,7 @@
 import java.util.stream.Collectors;
 
 @Slf4j
-public class GlobalStateProcessor <T> implements Processor<String, T> {
+public class GlobalStateProcessor <T> implements Processor<String, T, Void, Void> {
     private final String storeName;
     private final Consumer<List<T>> consumer;
     private KeyValueStore<String, T> store;
@@ -26,19 +27,20 @@ public GlobalStateProcessor(String storeName, Consumer<List<T>> consumer) {
         this.consumer = consumer;
     }
 
+    @SuppressWarnings("unchecked")
     @Override
     public void init(ProcessorContext context) {
-        this.store = context.getStateStore(this.storeName);
+        this.store = (KeyValueStore<String, T>) context.getStateStore(this.storeName);
 
         this.send();
     }
 
     @Override
-    public void process(String key, T value) {
-        if (value == null) {
-            this.store.delete(key);
+    public void process(Record<String, T> record) {
+        if (record.value() == null) {
+            this.store.delete(record.key());
         } else {
-            this.store.put(key, value);
+            this.store.put(record.key(), record.value());
         }
 
         this.send();
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/StateStoreTransformer.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/StateStoreTransformer.java
index f3a8a9fdde1..7a4d090f91b 100644
--- a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/StateStoreTransformer.java
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/StateStoreTransformer.java
@@ -19,9 +19,8 @@ public StateStoreTransformer(String storeName, Function<V, V> serializer) {
     }
 
     @Override
-    @SuppressWarnings("unchecked")
     public void init(final ProcessorContext context) {
-        this.store = (KeyValueStore<String, V>) context.getStateStore(this.storeName);
+        this.store = context.getStateStore(this.storeName);
     }
 
     @Override
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/WorkerInstanceTransformer.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/WorkerInstanceTransformer.java
index c9a4baa0a01..861cf1e756d 100644
--- a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/WorkerInstanceTransformer.java
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/WorkerInstanceTransformer.java
@@ -1,7 +1,10 @@
 package io.kestra.runner.kafka.streams;
 
 import com.google.common.collect.Streams;
-import io.kestra.core.runners.*;
+import io.kestra.core.runners.WorkerInstance;
+import io.kestra.core.runners.WorkerTask;
+import io.kestra.core.runners.WorkerTaskRunning;
+import io.kestra.core.services.WorkerInstanceService;
 import io.kestra.runner.kafka.executors.ExecutorWorkerRunning;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
@@ -13,14 +16,12 @@
 import org.apache.kafka.streams.state.KeyValueIterator;
 import org.apache.kafka.streams.state.KeyValueStore;
 import org.apache.kafka.streams.state.ValueAndTimestamp;
-import io.kestra.core.services.WorkerInstanceService;
 
 import java.util.Collections;
 import java.util.List;
 import java.util.stream.Collectors;
 
 @Slf4j
-@SuppressWarnings("UnstableApiUsage")
 public class WorkerInstanceTransformer implements ValueTransformerWithKey<String, WorkerInstance, List<WorkerInstanceTransformer.Result>> {
     private KeyValueStore<String, WorkerInstance> instanceStore;
     private KeyValueStore<String, ValueAndTimestamp<WorkerTaskRunning>> runningStore;
@@ -35,6 +36,7 @@ public void init(final ProcessorContext context) {
     }
 
     @Override
+    @SuppressWarnings("UnstableApiUsage")
     public List<Result> transform(final String key, final WorkerInstance value) {
         log.trace("Incoming instance: {} {}", key, value);
 
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/WorkerTaskExecutionTransformer.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/WorkerTaskExecutionTransformer.java
index 71c213b208d..61097a41844 100644
--- a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/WorkerTaskExecutionTransformer.java
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/WorkerTaskExecutionTransformer.java
@@ -5,7 +5,7 @@
 import io.kestra.core.runners.RunContextFactory;
 import io.kestra.core.runners.WorkerTaskExecution;
 import io.kestra.core.runners.WorkerTaskResult;
-import io.kestra.runner.kafka.services.SafeKeyValueStore;
+import io.kestra.runner.kafka.KafkaFlowExecutor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.streams.kstream.ValueTransformerWithKey;
 import org.apache.kafka.streams.processor.ProcessorContext;
@@ -18,21 +18,19 @@
 public class WorkerTaskExecutionTransformer implements ValueTransformerWithKey<String, Executor, WorkerTaskResult> {
     private final String workerTaskExecutionStoreName;
     private final RunContextFactory runContextFactory;
+    private final KafkaFlowExecutor kafkaFlowExecutor;
 
     private KeyValueStore<String, ValueAndTimestamp<WorkerTaskExecution>> workerTaskExecutionStore;
-    private SafeKeyValueStore<String, ValueAndTimestamp<Flow>> flowStore;
 
-    public WorkerTaskExecutionTransformer(RunContextFactory runContextFactory, String workerTaskExecutionStoreName) {
+    public WorkerTaskExecutionTransformer(RunContextFactory runContextFactory, String workerTaskExecutionStoreName, KafkaFlowExecutor kafkaFlowExecutor) {
         this.runContextFactory = runContextFactory;
         this.workerTaskExecutionStoreName = workerTaskExecutionStoreName;
+        this.kafkaFlowExecutor = kafkaFlowExecutor;
     }
 
     @Override
-    @SuppressWarnings("unchecked")
     public void init(final ProcessorContext context) {
-        var flowStore = (KeyValueStore<String, ValueAndTimestamp<Flow>>) context.getStateStore("flow");
-        this.flowStore = new SafeKeyValueStore<>(flowStore, flowStore.name());
-        this.workerTaskExecutionStore = (KeyValueStore<String, ValueAndTimestamp<WorkerTaskExecution>>) context.getStateStore(this.workerTaskExecutionStoreName);
+        this.workerTaskExecutionStore = context.getStateStore(this.workerTaskExecutionStoreName);
     }
 
     @Override
@@ -44,12 +42,12 @@ public WorkerTaskResult transform(final String key, final Executor value) {
 
         WorkerTaskExecution workerTaskExecution = workerTaskExecutionStoreValue.value();
 
-        Optional<ValueAndTimestamp<Flow>> flowValueAndTimestamp = this.flowStore.get(Flow.uid(value.getExecution()));
-        if (flowValueAndTimestamp.isEmpty()) {
+        Optional<Flow> flowOptional = this.kafkaFlowExecutor.findByExecution(value.getExecution());
+        if (flowOptional.isEmpty()) {
             return null;
         }
 
-        Flow flow = flowValueAndTimestamp.get().value();
+        Flow flow = flowOptional.get();
 
         return workerTaskExecution
             .getTask()
diff --git a/runner-kafka/src/test/java/io/kestra/runner/kafka/AbstractKafkaRunnerTest.java b/runner-kafka/src/test/java/io/kestra/runner/kafka/AbstractKafkaRunnerTest.java
index a43a68f0f7c..f245db5c8b6 100644
--- a/runner-kafka/src/test/java/io/kestra/runner/kafka/AbstractKafkaRunnerTest.java
+++ b/runner-kafka/src/test/java/io/kestra/runner/kafka/AbstractKafkaRunnerTest.java
@@ -1,15 +1,16 @@
 package io.kestra.runner.kafka;
 
-import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
-import org.junit.jupiter.api.BeforeEach;
 import io.kestra.core.repositories.LocalFlowRepositoryLoader;
 import io.kestra.core.runners.RunnerUtils;
 import io.kestra.core.runners.StandAloneRunner;
 import io.kestra.core.utils.TestsUtils;
+import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
+import jakarta.inject.Inject;
+import org.junit.jupiter.api.BeforeEach;
 
 import java.io.IOException;
 import java.net.URISyntaxException;
-import jakarta.inject.Inject;
+import java.util.List;
 
 @MicronautTest
 public abstract class AbstractKafkaRunnerTest {
@@ -22,8 +23,12 @@ public abstract class AbstractKafkaRunnerTest {
     @Inject
     protected LocalFlowRepositoryLoader repositoryLoader;
 
+    @Inject
+    protected KafkaTemplateExecutor kafkaTemplateExecutor;
+
     @BeforeEach
     private void init() throws IOException, URISyntaxException {
+        kafkaTemplateExecutor.setTemplates(List.of());
         TestsUtils.loads(repositoryLoader);
         runner.setSchedulerEnabled(false);
         runner.run();
diff --git a/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaExecutorTest.java b/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaExecutorTest.java
index b3e542732c7..601d02b07ab 100644
--- a/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaExecutorTest.java
+++ b/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaExecutorTest.java
@@ -17,14 +17,13 @@
 import io.kestra.core.utils.TestsUtils;
 import io.kestra.runner.kafka.configs.ClientConfig;
 import io.kestra.runner.kafka.configs.StreamDefaultsConfig;
-import io.kestra.runner.kafka.executors.ExecutorMain;
-import io.kestra.runner.kafka.executors.ExecutorFlowTrigger;
-import io.kestra.runner.kafka.executors.ExecutorWorkerRunning;
-import io.kestra.runner.kafka.executors.KafkaExecutorInterface;
+import io.kestra.runner.kafka.executors.*;
 import io.kestra.runner.kafka.serializers.JsonSerde;
 import io.kestra.runner.kafka.services.KafkaAdminService;
 import io.micronaut.context.ApplicationContext;
+import io.micronaut.context.annotation.Property;
 import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
+import jakarta.inject.Inject;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.common.serialization.StringSerializer;
@@ -40,7 +39,6 @@
 import java.io.IOException;
 import java.net.URISyntaxException;
 import java.util.*;
-import jakarta.inject.Inject;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.*;
@@ -48,6 +46,7 @@
 @SuppressWarnings("resource")
 @MicronautTest
 @Slf4j
+@Property(name = "kestra.server-type", value = "EXECUTOR")
 class KafkaExecutorTest {
     @Inject
     ApplicationContext applicationContext;
@@ -55,6 +54,9 @@ class KafkaExecutorTest {
     @Inject
     ExecutorMain executorMain;
 
+    @Inject
+    ExecutorStore executorStore;
+
     @Inject
     ExecutorFlowTrigger executorFlowTrigger;
 
@@ -76,10 +78,14 @@ class KafkaExecutorTest {
     @Inject
     FlowRepositoryInterface flowRepository;
 
+    @Inject
+    KafkaTemplateExecutor kafkaTemplateExecutor;
+
     TopologyTestDriver testTopology;
 
     @BeforeEach
     void init() throws IOException, URISyntaxException {
+        kafkaTemplateExecutor.setTemplates(List.of());
         TestsUtils.loads(repositoryLoader);
     }
 
@@ -104,13 +110,6 @@ void startStream(KafkaExecutorInterface kafkaExecutorInterface) {
         }
 
         testTopology = new TopologyTestDriver(topology, properties);
-
-        if (kafkaExecutorInterface.getClass() == ExecutorMain.class) {
-            applicationContext.registerSingleton(new KafkaTemplateExecutor(
-                testTopology.getKeyValueStore("template"),
-                "template"
-            ));
-        }
     }
 
     @AfterEach
@@ -122,11 +121,13 @@ void tear() {
 
     @Test
     void standard() {
-        startStream(this.executorMain);
+        startStream(this.executorStore);
 
         Flow flow = flowRepository.findById("io.kestra.tests", "logs").orElseThrow();
         this.flowInput().pipeInput(flow.uid(), flow);
 
+        startStream(this.executorMain);
+
         Execution execution = createExecution(flow);
 
         // task
@@ -152,11 +153,13 @@ void standard() {
 
     @Test
     void concurrent() {
-        startStream(this.executorMain);
+        startStream(this.executorStore);
 
         Flow flow = flowRepository.findById("io.kestra.tests", "logs").orElseThrow();
         this.flowInput().pipeInput(flow.uid(), flow);
 
+        startStream(this.executorMain);
+
         Execution execution = createExecution(flow);
 
         // task
@@ -183,11 +186,13 @@ void concurrent() {
 
     @Test
     void killed() {
-        startStream(this.executorMain);
+        startStream(this.executorStore);
 
         Flow flow = flowRepository.findById("io.kestra.tests", "logs").orElseThrow();
         this.flowInput().pipeInput(flow.uid(), flow);
 
+        startStream(this.executorMain);
+
         Execution execution = createExecution(flow);
 
         // task
@@ -224,11 +229,13 @@ void killed() {
 
     @Test
     void killedAlreadyFinished() {
-        startStream(this.executorMain);
+        startStream(this.executorStore);
 
         Flow flow = flowRepository.findById("io.kestra.tests", "logs").orElseThrow();
         this.flowInput().pipeInput(flow.uid(), flow);
 
+        startStream(this.executorMain);
+
         Execution execution = createExecution(flow);
 
         // task
@@ -247,11 +254,13 @@ void killedAlreadyFinished() {
     @ParameterizedTest
     @ValueSource(booleans = {true, false})
     void killedParallel(boolean killed) {
-        startStream(this.executorMain);
+        startStream(this.executorStore);
 
         Flow flow = flowRepository.findById("io.kestra.tests", "parallel").orElseThrow();
         this.flowInput().pipeInput(flow.uid(), flow);
 
+        startStream(this.executorMain);
+
         createExecution(flow);
         Parallel parent = (Parallel) flow.getTasks().get(0);
         Task firstChild = parent.getTasks().get(0);
@@ -314,11 +323,13 @@ void killedParallel(boolean killed) {
 
     @Test
     void eachNull() {
-        startStream(this.executorMain);
+        startStream(this.executorStore);
 
         Flow flow = flowRepository.findById("io.kestra.tests", "each-null").orElseThrow();
         this.flowInput().pipeInput(flow.uid(), flow);
 
+        startStream(this.executorMain);
+
         Execution execution = createExecution(flow);
         execution = executionOutput().readRecord().value();
         execution = executionOutput().readRecord().value();
@@ -331,11 +342,13 @@ void eachNull() {
 
     @Test
     void parallel() {
-        startStream(this.executorMain);
+        startStream(this.executorStore);
 
         Flow flow = flowRepository.findById("io.kestra.tests", "parallel").orElseThrow();
         this.flowInput().pipeInput(flow.uid(), flow);
 
+        startStream(this.executorMain);
+
         createExecution(flow);
         Parallel parent = (Parallel) flow.getTasks().get(0);
         Task last = flow.getTasks().get(1);
@@ -396,11 +409,13 @@ void parallel() {
 
     @Test
     void eachParallelNested() throws InternalException {
-        startStream(this.executorMain);
+        startStream(this.executorStore);
 
         Flow flow = flowRepository.findById("io.kestra.tests", "each-parallel-nested").orElseThrow();
         this.flowInput().pipeInput(flow.uid(), flow);
 
+        startStream(this.executorMain);
+
         Execution execution = createExecution(flow);
 
         for (int i = 0; i < 5; i++) {
@@ -454,7 +469,7 @@ void eachParallelNested() throws InternalException {
 
     @Test
     void flowTrigger() {
-        startStream(this.executorMain);
+        startStream(this.executorStore);
 
         Flow triggerFlow = flowRepository.findById("io.kestra.tests", "trigger-flow-listener-no-inputs").orElseThrow();
         this.flowInput().pipeInput(triggerFlow.uid(), triggerFlow);
@@ -466,6 +481,8 @@ void flowTrigger() {
         Flow firstFlow = flowRepository.findById("io.kestra.tests", "trigger-flow").orElseThrow();
         this.flowInput().pipeInput(firstFlow.uid(), firstFlow);
 
+        startStream(this.executorMain);
+
         Execution firstExecution = createExecution(firstFlow);
 
         // task
@@ -486,7 +503,7 @@ void flowTrigger() {
 
     @Test
     void multipleTrigger() {
-        startStream(this.executorMain);
+        startStream(this.executorStore);
 
         Flow flowA = flowRepository.findById("io.kestra.tests", "trigger-multiplecondition-flow-a").orElseThrow();
         this.flowInput().pipeInput(flowA.uid(), flowA);
@@ -497,6 +514,12 @@ void multipleTrigger() {
         Flow triggerFlow = flowRepository.findById("io.kestra.tests", "trigger-multiplecondition-listener").orElseThrow();
         this.flowInput().pipeInput(triggerFlow.uid(), triggerFlow);
 
+        this.flowInput().pipeInput(flowA.uid(), flowA);
+        this.flowInput().pipeInput(flowB.uid(), flowB);
+        this.flowInput().pipeInput(triggerFlow.uid(), triggerFlow);
+
+        startStream(this.executorMain);
+
         // first
         Execution executionA = createExecution(flowA);
         executionA = runningAndSuccessSequential(flowA, executionA, 0);
@@ -514,9 +537,6 @@ void multipleTrigger() {
         assertThat(executorFlowTriggerB.getExecution().getFlowId(), is("trigger-multiplecondition-flow-b"));
 
         startStream(this.executorFlowTrigger);
-        this.flowInput().pipeInput(flowA.uid(), flowA);
-        this.flowInput().pipeInput(flowB.uid(), flowB);
-        this.flowInput().pipeInput(triggerFlow.uid(), triggerFlow);
 
         executorFlowTriggerInput().pipeInput(executorFlowTriggerA.getFlowHavingTrigger().uid(), executorFlowTriggerA);
         executorFlowTriggerInput().pipeInput(executorFlowTriggerB.getFlowHavingTrigger().uid(), executorFlowTriggerB);
@@ -529,12 +549,14 @@ void multipleTrigger() {
 
     @Test
     void workerRebalanced() {
-        startStream(this.executorMain);
+        startStream(this.executorStore);
 
         // start an execution to generate some id
         Flow flow = flowRepository.findById("io.kestra.tests", "logs").orElseThrow();
         this.flowInput().pipeInput(flow.uid(), flow);
 
+        startStream(this.executorMain);
+
         Execution execution = createExecution(flow);
 
         execution = runningAndSuccessSequential(flow, execution, 0, State.Type.RUNNING);
@@ -574,7 +596,7 @@ void workerRebalanced() {
 
     @Test
     void invalidStore() throws JsonProcessingException {
-        startStream(this.executorMain);
+        startStream(this.executorStore);
 
         KeyValueStore<String, String> flow = this.testTopology.getKeyValueStore("flow");
         flow.put("io.kestra.unittest_invalid_1", JacksonMapper.ofJson().writeValueAsString(Map.of(
@@ -596,6 +618,8 @@ void invalidStore() throws JsonProcessingException {
         Flow firstFlow = flowRepository.findById("io.kestra.tests", "trigger-flow").orElseThrow();
         this.flowInput().pipeInput(firstFlow.uid(), firstFlow);
 
+        startStream(this.executorMain);
+
         Execution firstExecution = createExecution(firstFlow);
 
         // task
diff --git a/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaFlowListenersTest.java b/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaFlowListenersTest.java
index e417c03377b..01c1e481d78 100644
--- a/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaFlowListenersTest.java
+++ b/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaFlowListenersTest.java
@@ -9,6 +9,7 @@
 import io.kestra.runner.kafka.configs.TopicsConfig;
 import io.kestra.runner.kafka.services.KafkaProducerService;
 import io.micronaut.context.ApplicationContext;
+import io.micronaut.context.annotation.Property;
 import jakarta.inject.Inject;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.serialization.Serdes;
@@ -23,6 +24,7 @@
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
 
+@Property(name = "kestra.server-type", value = "EXECUTOR")
 class KafkaFlowListenersTest extends FlowListenersTest {
     @Inject
     ApplicationContext applicationContext;
diff --git a/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaRunnerTest.java b/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaRunnerTest.java
index 4ebf459ad7d..ade1087d249 100644
--- a/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaRunnerTest.java
+++ b/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaRunnerTest.java
@@ -13,6 +13,9 @@
 import io.kestra.core.tasks.flows.FlowCaseTest;
 import io.kestra.core.tasks.flows.TemplateTest;
 import io.kestra.core.utils.TestsUtils;
+import io.micronaut.context.annotation.Property;
+import jakarta.inject.Inject;
+import jakarta.inject.Named;
 import org.apache.kafka.common.errors.RecordTooLargeException;
 import org.junit.jupiter.api.Test;
 
@@ -22,13 +25,12 @@
 import java.util.*;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeoutException;
-import jakarta.inject.Inject;
-import jakarta.inject.Named;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.*;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
+@Property(name = "kestra.server-type", value = "EXECUTOR")
 class KafkaRunnerTest extends AbstractKafkaRunnerTest {
     @Inject
     private RestartCaseTest restartCaseTest;
diff --git a/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaSchedulerTest.java b/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaSchedulerTest.java
index 56f865ea7dd..5da92d7a001 100644
--- a/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaSchedulerTest.java
+++ b/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaSchedulerTest.java
@@ -13,6 +13,7 @@
 import io.kestra.runner.kafka.serializers.JsonSerde;
 import io.kestra.runner.kafka.services.KafkaAdminService;
 import io.kestra.runner.kafka.services.KafkaProducerService;
+import io.micronaut.context.annotation.Property;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.jupiter.api.BeforeEach;
@@ -26,6 +27,7 @@
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
 
+@Property(name = "kestra.server-type", value = "EXECUTOR")
 class KafkaSchedulerTest extends AbstractSchedulerTest {
     @Inject
     protected KafkaFlowListeners flowListeners;
diff --git a/runner-kafka/src/test/java/io/kestra/runner/kafka/services/SafeKeyValueStoreTest.java b/runner-kafka/src/test/java/io/kestra/runner/kafka/services/SafeKeyValueStoreTest.java
index 3d4dc139095..6d1c5e6b8c7 100644
--- a/runner-kafka/src/test/java/io/kestra/runner/kafka/services/SafeKeyValueStoreTest.java
+++ b/runner-kafka/src/test/java/io/kestra/runner/kafka/services/SafeKeyValueStoreTest.java
@@ -4,11 +4,17 @@
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.serializers.JacksonMapper;
 import io.kestra.core.tasks.debugs.Echo;
+import io.kestra.runner.kafka.serializers.JsonSerde;
 import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
+import jakarta.inject.Inject;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.utils.Bytes;
 import org.apache.kafka.streams.StreamsBuilder;
 import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.Topology;
 import org.apache.kafka.streams.TopologyTestDriver;
+import org.apache.kafka.streams.kstream.Consumed;
+import org.apache.kafka.streams.kstream.Materialized;
 import org.apache.kafka.streams.state.KeyValueStore;
 import org.junit.jupiter.api.Test;
 
@@ -17,7 +23,6 @@
 import java.util.Optional;
 import java.util.Properties;
 import java.util.stream.Collectors;
-import jakarta.inject.Inject;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
@@ -38,7 +43,14 @@ void test() throws JsonProcessingException {
         properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "unit-test");
 
         StreamsBuilder builder = new KafkaStreamsBuilder();
-        kafkaStreamSourceService.flowGlobalKTable(builder);
+        builder
+            .globalTable(
+                kafkaAdminService.getTopicName(Flow.class),
+                Consumed.with(Serdes.String(), JsonSerde.of(Flow.class)).withName("GlobalKTable.Flow"),
+                Materialized.<String, Flow, KeyValueStore<Bytes, byte[]>>as("flow")
+                    .withKeySerde(Serdes.String())
+                    .withValueSerde(JsonSerde.of(Flow.class))
+            );
 
         Topology topology = builder.build();
 
diff --git a/runner-memory/src/main/java/io/kestra/runner/memory/MemoryExecutor.java b/runner-memory/src/main/java/io/kestra/runner/memory/MemoryExecutor.java
index deaade72f83..00d85514ceb 100644
--- a/runner-memory/src/main/java/io/kestra/runner/memory/MemoryExecutor.java
+++ b/runner-memory/src/main/java/io/kestra/runner/memory/MemoryExecutor.java
@@ -100,7 +100,7 @@ private Flow transform(Flow flow, Execution execution) {
             flow = Template.injectTemplate(
                 flow,
                 execution,
-                templateExecutorInterface::findById
+                (namespace, id) -> templateExecutorInterface.findById(namespace, id).orElse(null)
             );
         } catch (InternalException e) {
             log.warn("Failed to inject template",  e);