fix(kafka-runner): add a Await on FlowJoinerTransformer

if the flow is already fetch, this crashed the executor since before we are sending null values

runner-kafka/src/main/java/io/kestra/runner/kafka/streams/FlowJoinerTransformer.java

@@ -6,10 +6,13 @@
 import io.kestra.core.runners.FlowExecutorInterface;
 import io.kestra.core.services.TaskDefaultService;
 import io.kestra.core.tasks.flows.Template;
+import io.kestra.core.utils.Await;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.streams.kstream.ValueTransformerWithKey;
 import org.apache.kafka.streams.processor.ProcessorContext;
 
+import java.time.Duration;
+import java.time.Instant;
 import java.util.Optional;
 
 @Slf4j
@@ -33,13 +36,24 @@ public void init(final ProcessorContext context) {
 
     @Override
     public Executor transform(String key, Executor executor) {
-        Optional<Flow> flowState = flowExecutorInterface.findByExecution(executor.getExecution());
+        Instant start = Instant.now();
 
-        if (flowState.isEmpty()) {
-            return null;
-        }
+        // pooling of new flow can be delayed on ExecutorStore, we maybe need to wait that the flow is updated
+        Flow flow = Await.until(
+            () -> {
+                Optional<Flow> flowState = flowExecutorInterface.findByExecution(executor.getExecution());
+
+                if (flowState.isEmpty() && start.plusSeconds(60).isBefore(Instant.now())) {
+                    log.warn("Unable to find flow with namespace: '" + executor.getExecution().getNamespace() +
+                        ", id: " + executor.getExecution().getFlowId() + "', " +
+                        "revision '" + executor.getExecution().getFlowRevision() + "' " +
+                        "since '" + start  + "'");
+                }
 
-        Flow flow = flowState.get();
+                return flowState.orElse(null);
+            },
+            Duration.ofMillis(100)
+        );
 
         if (!withDefaults) {
             return executor.withFlow(flow);