Repository: kestra-io/kestra
Issue: Remove trigger last execution when we remove the trigger
Labels: bug
backend

When the user remove a trigger the last trigger is not removed.
If the user recreate the same trigger, the previous value is kept and the backfill will start previous date. 
Just delete the lastrigger on this case to reset all previous job 

Issue comments:

Pull request review comments:

Patch that resolved the issue:
diff --git a/core/src/main/java/org/kestra/core/models/triggers/Trigger.java b/core/src/main/java/org/kestra/core/models/triggers/Trigger.java
index ab54eddc00..86aa1599b2 100644
--- a/core/src/main/java/org/kestra/core/models/triggers/Trigger.java
+++ b/core/src/main/java/org/kestra/core/models/triggers/Trigger.java
@@ -3,6 +3,7 @@
 import lombok.*;
 import lombok.experimental.SuperBuilder;
 import org.kestra.core.models.executions.Execution;
+import org.kestra.core.models.flows.Flow;
 
 import java.util.Arrays;
 import javax.validation.constraints.NotNull;
@@ -36,6 +37,15 @@ public static String uid(Execution execution) {
         ));
     }
 
+    public static Trigger of(Flow flow, AbstractTrigger abstractTrigger) {
+        return Trigger.builder()
+            .namespace(flow.getNamespace())
+            .flowId(flow.getId())
+            .flowRevision(flow.getRevision())
+            .triggerId(abstractTrigger.getId())
+            .build();
+    }
+
     public static Trigger of(TriggerContext triggerContext, Execution execution) {
         return Trigger.builder()
             .namespace(triggerContext.getNamespace())
diff --git a/core/src/main/java/org/kestra/core/queues/QueueInterface.java b/core/src/main/java/org/kestra/core/queues/QueueInterface.java
index 9af797129b..213b9ffa26 100644
--- a/core/src/main/java/org/kestra/core/queues/QueueInterface.java
+++ b/core/src/main/java/org/kestra/core/queues/QueueInterface.java
@@ -6,6 +6,8 @@
 public interface QueueInterface<T> extends Closeable {
     void emit(T message) throws QueueException;
 
+    void delete(T message) throws QueueException;
+
     Runnable receive(Consumer<T> consumer);
 
     Runnable receive(Class<?> consumerGroup, Consumer<T> consumer);
diff --git a/core/src/main/java/org/kestra/core/services/FlowService.java b/core/src/main/java/org/kestra/core/services/FlowService.java
index f5f9823fa4..9f00ab6213 100644
--- a/core/src/main/java/org/kestra/core/services/FlowService.java
+++ b/core/src/main/java/org/kestra/core/services/FlowService.java
@@ -7,6 +7,7 @@
 import org.kestra.core.models.flows.Flow;
 import org.kestra.core.models.triggers.AbstractTrigger;
 import org.kestra.core.runners.RunContextFactory;
+import org.kestra.core.utils.ListUtils;
 
 import java.util.*;
 import java.util.stream.Collectors;
@@ -85,6 +86,16 @@ public List<Execution> flowTriggerExecution(Stream<Flow> flowStream, Execution e
             .collect(Collectors.toList());
     }
 
+    public static List<AbstractTrigger> findRemovedTrigger(Flow flow, Flow previous) {
+        return ListUtils.emptyOnNull(previous.getTriggers())
+            .stream()
+            .filter(p -> ListUtils.emptyOnNull(flow.getTriggers())
+                .stream()
+                .noneMatch(c -> c.getId().equals(p.getId()))
+            )
+            .collect(Collectors.toList());
+    }
+
     @AllArgsConstructor
     @Getter
     private static class FlowWithTrigger {
diff --git a/core/src/main/java/org/kestra/core/utils/ListUtils.java b/core/src/main/java/org/kestra/core/utils/ListUtils.java
new file mode 100644
index 0000000000..393c3bb441
--- /dev/null
+++ b/core/src/main/java/org/kestra/core/utils/ListUtils.java
@@ -0,0 +1,10 @@
+package org.kestra.core.utils;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class ListUtils {
+    public static <T> List<T> emptyOnNull(List<T> list) {
+        return list == null ? new ArrayList<>() : list;
+    }
+}
diff --git a/indexer-kafka-elasticsearch/src/main/java/org/kestra/indexer/KafkaElasticIndexer.java b/indexer-kafka-elasticsearch/src/main/java/org/kestra/indexer/KafkaElasticIndexer.java
index bea0958fed..42655141b1 100644
--- a/indexer-kafka-elasticsearch/src/main/java/org/kestra/indexer/KafkaElasticIndexer.java
+++ b/indexer-kafka-elasticsearch/src/main/java/org/kestra/indexer/KafkaElasticIndexer.java
@@ -11,6 +11,7 @@
 import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.delete.DeleteRequest;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.RestHighLevelClient;
@@ -151,11 +152,20 @@ private BulkRequest bulkRequest(List<ConsumerRecord<String, String>> rows) {
         BulkRequest request = new BulkRequest();
 
         rows
-            .forEach(record -> request
-                .add(new IndexRequest(this.indexName(record))
-                .id(record.key())
-                .source(record.value(), XContentType.JSON)
-            ));
+            .forEach(record -> {
+                if (record.value() == null) {
+                    request
+                        .add(new DeleteRequest(this.indexName(record))
+                            .id(record.key())
+                        );
+                } else {
+                    request
+                        .add(new IndexRequest(this.indexName(record))
+                            .id(record.key())
+                            .source(record.value(), XContentType.JSON)
+                        );
+                }
+            });
 
         return request;
     }
diff --git a/repository-elasticsearch/src/main/java/org/kestra/repository/elasticsearch/ElasticSearchFlowRepository.java b/repository-elasticsearch/src/main/java/org/kestra/repository/elasticsearch/ElasticSearchFlowRepository.java
index 40f11f65a3..c76c691c0d 100644
--- a/repository-elasticsearch/src/main/java/org/kestra/repository/elasticsearch/ElasticSearchFlowRepository.java
+++ b/repository-elasticsearch/src/main/java/org/kestra/repository/elasticsearch/ElasticSearchFlowRepository.java
@@ -8,11 +8,13 @@
 import org.elasticsearch.search.sort.FieldSortBuilder;
 import org.elasticsearch.search.sort.SortOrder;
 import org.kestra.core.models.flows.Flow;
+import org.kestra.core.models.triggers.Trigger;
 import org.kestra.core.models.validations.ModelValidator;
 import org.kestra.core.queues.QueueFactoryInterface;
 import org.kestra.core.queues.QueueInterface;
 import org.kestra.core.repositories.ArrayListTotal;
 import org.kestra.core.repositories.FlowRepositoryInterface;
+import org.kestra.core.services.FlowService;
 import org.kestra.core.utils.ExecutorsUtils;
 import org.kestra.repository.elasticsearch.configs.IndicesConfig;
 
@@ -31,6 +33,7 @@ public class ElasticSearchFlowRepository extends AbstractElasticSearchRepository
     protected static final String REVISIONS_NAME = "flows-revisions";
 
     private final QueueInterface<Flow> flowQueue;
+    private final QueueInterface<Trigger> triggerQueue;
 
     @Inject
     public ElasticSearchFlowRepository(
@@ -38,11 +41,13 @@ public ElasticSearchFlowRepository(
         List<IndicesConfig> indicesConfigs,
         ModelValidator modelValidator,
         ExecutorsUtils executorsUtils,
-        @Named(QueueFactoryInterface.FLOW_NAMED) QueueInterface<Flow> flowQueue
+        @Named(QueueFactoryInterface.FLOW_NAMED) QueueInterface<Flow> flowQueue,
+        @Named(QueueFactoryInterface.TRIGGER_NAMED) QueueInterface<Trigger> triggerQueue
     ) {
         super(client, indicesConfigs, modelValidator, executorsUtils, Flow.class);
 
         this.flowQueue = flowQueue;
+        this.triggerQueue = triggerQueue;
     }
 
     private static String flowId(Flow flow) {
@@ -129,7 +134,13 @@ public Flow update(Flow flow, Flow previous) throws ConstraintViolationException
                 throw s;
             });
 
-        return this.save(flow);
+        Flow saved = this.save(flow);
+
+        FlowService
+            .findRemovedTrigger(flow, previous)
+            .forEach(abstractTrigger -> triggerQueue.delete(Trigger.of(flow, abstractTrigger)));
+
+        return saved;
     }
 
     public Flow save(Flow flow) throws ConstraintViolationException {
diff --git a/repository-memory/src/main/java/org/kestra/repository/memory/MemoryFlowRepository.java b/repository-memory/src/main/java/org/kestra/repository/memory/MemoryFlowRepository.java
index 8ed63faed4..dade469763 100644
--- a/repository-memory/src/main/java/org/kestra/repository/memory/MemoryFlowRepository.java
+++ b/repository-memory/src/main/java/org/kestra/repository/memory/MemoryFlowRepository.java
@@ -2,12 +2,14 @@
 
 import io.micronaut.core.value.ValueException;
 import io.micronaut.data.model.Pageable;
+import org.kestra.core.models.triggers.Trigger;
 import org.kestra.core.models.validations.ModelValidator;
 import org.kestra.core.models.flows.Flow;
 import org.kestra.core.queues.QueueFactoryInterface;
 import org.kestra.core.queues.QueueInterface;
 import org.kestra.core.repositories.ArrayListTotal;
 import org.kestra.core.repositories.FlowRepositoryInterface;
+import org.kestra.core.services.FlowService;
 
 import javax.inject.Inject;
 import javax.inject.Named;
@@ -26,6 +28,10 @@ public class MemoryFlowRepository implements FlowRepositoryInterface {
     @Named(QueueFactoryInterface.FLOW_NAMED)
     private QueueInterface<Flow> flowQueue;
 
+    @Inject
+    @Named(QueueFactoryInterface.TRIGGER_NAMED)
+    private QueueInterface<Trigger> triggerQueue;
+
     @Inject
     private ModelValidator modelValidator;
 
@@ -106,7 +112,13 @@ public Flow update(Flow flow, Flow previous) throws ConstraintViolationException
                 throw s;
             });
 
-        return this.save(flow);
+        Flow saved = this.save(flow);
+
+        FlowService
+            .findRemovedTrigger(flow, previous)
+            .forEach(abstractTrigger -> triggerQueue.delete(Trigger.of(flow, abstractTrigger)));
+
+        return saved;
     }
 
     private Flow save(Flow flow) throws ConstraintViolationException {
diff --git a/runner-kafka/src/main/java/org/kestra/runner/kafka/KafkaQueue.java b/runner-kafka/src/main/java/org/kestra/runner/kafka/KafkaQueue.java
index 97c3a1e538..e50c4b4d03 100644
--- a/runner-kafka/src/main/java/org/kestra/runner/kafka/KafkaQueue.java
+++ b/runner-kafka/src/main/java/org/kestra/runner/kafka/KafkaQueue.java
@@ -85,15 +85,16 @@ static <T> void log(TopicsConfig topicsConfig, T object, String message) {
         }
     }
 
-    @Override
-    public void emit(T message) throws QueueException {
+    private void produce(String key, T message) {
         KafkaQueue.log(topicsConfig, message, "Outgoing messsage");
 
         try {
             kafkaProducer
-                .send(new ProducerRecord<>(
-                    topicsConfig.getName(),
-                    key(message), message),
+                .send(
+                    new ProducerRecord<>(
+                        topicsConfig.getName(),
+                        key, message
+                    ),
                     (metadata, e) -> {
                         if (e != null) {
                             log.error("Failed to produce '{}' with metadata '{}'", e, metadata);
@@ -106,6 +107,17 @@ public void emit(T message) throws QueueException {
         }
     }
 
+    @Override
+    public void emit(T message) throws QueueException {
+        this.produce(key(message), message);
+    }
+
+
+    @Override
+    public void delete(T message) throws QueueException {
+        this.produce(key(message), null);
+    }
+
     @Override
     public Runnable receive(Consumer<T> consumer) {
         return this.receive(null, consumer);
diff --git a/runner-memory/src/main/java/org/kestra/runner/memory/MemoryQueue.java b/runner-memory/src/main/java/org/kestra/runner/memory/MemoryQueue.java
index 44141f09c7..1befb13abc 100644
--- a/runner-memory/src/main/java/org/kestra/runner/memory/MemoryQueue.java
+++ b/runner-memory/src/main/java/org/kestra/runner/memory/MemoryQueue.java
@@ -6,6 +6,7 @@
 import io.micronaut.context.ApplicationContext;
 import lombok.extern.slf4j.Slf4j;
 import org.kestra.core.queues.AbstractQueue;
+import org.kestra.core.queues.QueueException;
 import org.kestra.core.queues.QueueInterface;
 import org.kestra.core.serializers.JacksonMapper;
 import org.kestra.core.utils.ExecutorsUtils;
@@ -43,8 +44,7 @@ private static int selectConsumer(String key, int size) {
         }
     }
 
-    @Override
-    public void emit(T message) {
+    private void produce(String key, T message) {
         if (log.isTraceEnabled()) {
             log.trace("New message: topic '{}', value {}", this.cls.getName(), message);
         }
@@ -59,7 +59,7 @@ public void emit(T message) {
                             log.debug("No consumer connected on queue '" + this.cls.getName() + "'");
                             return;
                         } else {
-                            int index = selectConsumer(key(message), consumers.size());
+                            int index = selectConsumer(key, consumers.size());
                             consumer = consumers.get(index);
                         }
                     }
@@ -67,7 +67,7 @@ public void emit(T message) {
                     // we force serialization to be a the same case than an another queue with serialization
                     // this enabled debugging classLoader
                     try {
-                        T serialized = mapper.readValue(mapper.writeValueAsString(message), this.cls);
+                        T serialized = message == null ? null : mapper.readValue(mapper.writeValueAsString(message), this.cls);
                         consumer.accept(serialized);
                     } catch (JsonProcessingException e) {
                         throw new RuntimeException(e);
@@ -76,6 +76,16 @@ public void emit(T message) {
             });
     }
 
+    @Override
+    public void emit(T message) {
+        this.produce(key(message), message);
+    }
+
+    @Override
+    public void delete(T message) throws QueueException {
+        this.produce(key(message), null);
+    }
+
     @Override
     public Runnable receive(Consumer<T> consumer) {
         return this.receive(null, consumer);


Test patch:
diff --git a/core/src/test/java/org/kestra/core/repositories/AbstractFlowRepositoryTest.java b/core/src/test/java/org/kestra/core/repositories/AbstractFlowRepositoryTest.java
index 802ac43880..1e78e978f5 100644
--- a/core/src/test/java/org/kestra/core/repositories/AbstractFlowRepositoryTest.java
+++ b/core/src/test/java/org/kestra/core/repositories/AbstractFlowRepositoryTest.java
@@ -8,6 +8,10 @@
 import org.kestra.core.Helpers;
 import org.kestra.core.models.flows.Flow;
 import org.kestra.core.models.flows.Input;
+import org.kestra.core.models.triggers.Trigger;
+import org.kestra.core.queues.QueueFactoryInterface;
+import org.kestra.core.queues.QueueInterface;
+import org.kestra.core.schedulers.AbstractSchedulerTest;
 import org.kestra.core.serializers.JacksonMapper;
 import org.kestra.core.tasks.debugs.Return;
 import org.kestra.core.tasks.scripts.Bash;
@@ -19,11 +23,14 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
 import javax.inject.Inject;
+import javax.inject.Named;
 import javax.validation.ConstraintViolationException;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.nullValue;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
 @MicronautTest
@@ -34,6 +41,10 @@ public abstract class AbstractFlowRepositoryTest {
     @Inject
     private LocalFlowRepositoryLoader repositoryLoader;
 
+    @Inject
+    @Named(QueueFactoryInterface.TRIGGER_NAMED)
+    private QueueInterface<Trigger> triggerQueue;
+
     @BeforeEach
     private void init() throws IOException, URISyntaxException {
         TestsUtils.loads(repositoryLoader);
@@ -172,7 +183,6 @@ void delete() {
         assertThat(flowRepository.findById(flow.getNamespace(), flow.getId()).isPresent(), is(false));
     }
 
-
     @Test
     void updateConflict() {
         String flowId = IdUtils.create();
@@ -203,6 +213,45 @@ void updateConflict() {
         assertThat(e.getConstraintViolations().size(), is(2));
 
         flowRepository.delete(save);
+    }
+
+    @Test
+    void removeTrigger() throws InterruptedException {
+        CountDownLatch countDownLatch = new CountDownLatch(1);
+
+        triggerQueue.receive(trigger -> {
+            assertThat(trigger, is(nullValue()));
+            countDownLatch.countDown();
+        });
+
+        String flowId = IdUtils.create();
+
+        Flow flow = Flow.builder()
+            .id(flowId)
+            .namespace("org.kestra.unittest")
+            .triggers(Collections.singletonList(AbstractSchedulerTest.UnitTest.builder()
+                .id("sleep")
+                .type(AbstractSchedulerTest.UnitTest.class.getName())
+                .build()))
+            .tasks(Collections.singletonList(Return.builder().id("test").type(Return.class.getName()).format("test").build()))
+            .build();
+
+        Flow save = flowRepository.create(flow);
+
+        assertThat(flowRepository.findById(flow.getNamespace(), flow.getId()).isPresent(), is(true));
 
+        Flow update = Flow.builder()
+            .id(flowId)
+            .namespace("org.kestra.unittest")
+            .tasks(Collections.singletonList(Return.builder().id("test").type(Return.class.getName()).format("test").build()))
+            .build();;
+
+        Flow updated = flowRepository.update(update, flow);
+
+        countDownLatch.await();
+
+        assertThat(updated.getTriggers(), is(nullValue()));
+
+        flowRepository.delete(save);
     }
 }
diff --git a/indexer-kafka-elasticsearch/src/test/java/org/kestra/indexer/KafkaElasticIndexerTest.java b/indexer-kafka-elasticsearch/src/test/java/org/kestra/indexer/KafkaElasticIndexerTest.java
index 7547691739..63c7e539e3 100644
--- a/indexer-kafka-elasticsearch/src/test/java/org/kestra/indexer/KafkaElasticIndexerTest.java
+++ b/indexer-kafka-elasticsearch/src/test/java/org/kestra/indexer/KafkaElasticIndexerTest.java
@@ -84,11 +84,14 @@ void run() throws IOException, InterruptedException {
         MockConsumer<String, String> mockConsumer = mockConsumer(topic);
         doReturn(mockConsumer).when(kafkaConsumerServiceSpy).of(any(), any());
 
-        mockConsumer.addRecord(buildExecutionRecord(topic, 0));
+        ConsumerRecord<String, String> first = buildExecutionRecord(topic, 0);
+
+        mockConsumer.addRecord(first);
         mockConsumer.addRecord(buildExecutionRecord(topic, 1));
         mockConsumer.addRecord(buildExecutionRecord(topic, 2));
         mockConsumer.addRecord(buildExecutionRecord(topic, 3));
         mockConsumer.addRecord(buildExecutionRecord(topic, 4));
+        mockConsumer.addRecord(buildRecord(topic, first.key(), null, 5));
 
         KafkaElasticIndexer indexer = new KafkaElasticIndexer(
             metricRegistry,
@@ -112,6 +115,7 @@ void run() throws IOException, InterruptedException {
     private ConsumerRecord<String, String> buildExecutionRecord(String topic, int offset) throws JsonProcessingException {
         Flow flow = TestsUtils.mockFlow();
         Execution execution = TestsUtils.mockExecution(flow, ImmutableMap.of());
+
         return buildRecord(
             topic,
             execution.getId(),