fix(kafka-runner): restart & change status of an execution don't laun…

…ch depending flows close #607
kestra-io · Jun 23, 2022 · 4a61af4 · 4a61af4
1 parent c4d6cca
commit 4a61af4
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 4 deletions.
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
@@ -149,7 +149,7 @@ public StreamsBuilder topology() {
         this.handleWorkerTaskExecution(workerTaskExecutionKTable, stream);
 
         // purge at end
-        this.purgeExecutor(stream);
+        this.purgeExecutor(executorKStream);
 
         this.purgeWorkerRunning(workerTaskResultKStream);
 
@@ -332,6 +332,17 @@ private void purgeExecutor(KStream<String, Executor> stream) {
                 NEXTS_DEDUPLICATION_STATE_STORE_NAME
             );
 
+        // clean up Flow Trigger deduplication state
+        terminated
+            .transformValues(
+                () -> new DeduplicationPurgeTransformer<>(
+                    TRIGGER_DEDUPLICATION_STATE_STORE_NAME,
+                    (key, value) -> value.getExecution().getId()
+                ),
+                Named.as("PurgeExecutor.purgeTriggerDeduplication"),
+                TRIGGER_DEDUPLICATION_STATE_STORE_NAME
+            );
+
         // clean up killed
         terminated
             .filter(

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
@@ -144,11 +144,9 @@ void standard() {
         assertThat(workerTaskRunningOutput().readRecord().value(), is(nullValue()));
         assertThat(workerTaskRunningOutput().isEmpty(), is(true));
 
-        // executor topic must be deleted @TODO: 2 null values
+        // executor topic must be deleted
         TestRecord<String, Executor> executor = executorOutput().readRecord();
         assertThat(executor.value(), is(nullValue()));
-        executor = executorOutput().readRecord();
-        assertThat(executor.value(), is(nullValue()));
         assertThat(executorOutput().isEmpty(), is(true));
     }