Repository: kestra-io/kestra (java)

Issue:
Add a generic `stopAfter` property allowing to disable any trigger after a SUCCESS (e.g. HTTP trigger condition) or FAILED (e.g. dependsOnPast behavior) execution
### Feature description

Example syntax:

```yaml
id: hello-world
namespace: dev

tasks:
 - id: hello
   type: io.kestra.core.tasks.log.Log
   message: pausing a trigger after a success

triggers:
 - id: stopAfter
   type: io.kestra.core.models.triggers.types.Schedule
   cron: "0 9 * * *"
   stopAfter: # list of enum-type Execution states - can be set to any Execution states incl. multiple states as it's a list property, similarly to how ExecutionState condition on a Flow trigger accepts multiple possible states
     - SUCCESS # HTTP trigger as exist condition
     - FAILED # dependsOnPast for past Airflow users who want to disable the  executions until a broken pipeline gets fixed
```

**Important notes:**
- Ideally, disabling a trigger should generate a dedicated Audit log in EE e.g. "Trigger state action" with info which process/human disabled the trigger. For now, this doesn't seem feasible; see the comment below.
- This property is meant to be used primarily for a Schedule trigger + triggers with conditions incl. HTTP, JDBC, File Detection triggers; we will add it as a generic property for all triggers to stay flexible but just a note that for message queues, wehook or Flow triggers, this property is less relevant.
- We don't handle any automatic reenabling logic. After a trigger has been disabled, this generates a specific audit log and the user can then take some action based on it e.g. as part of https://github.com/kestra-io/kestra-ee/issues/809 

Pull request description:
feat(core): Trigger stopAfter
Fixes #2964

Example flow:

```yaml
id: hello-schedule2
namespace: company.team

triggers:
  - id: schedule
    type: io.kestra.core.models.triggers.types.Schedule
    cron: "* * * * *"
    stopAfter: 
      - SUCCESS

tasks:
  - id: hello
    type: io.kestra.core.tasks.log.Log
    message: Kestra team wishes you a great day! 👋
```

Patch:
diff --git a/core/src/main/java/io/kestra/core/models/triggers/AbstractTrigger.java b/core/src/main/java/io/kestra/core/models/triggers/AbstractTrigger.java
index dd156d0d0..d20c68992 100644
--- a/core/src/main/java/io/kestra/core/models/triggers/AbstractTrigger.java
+++ b/core/src/main/java/io/kestra/core/models/triggers/AbstractTrigger.java
@@ -4,6 +4,7 @@ import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import io.kestra.core.models.annotations.PluginProperty;
 import io.kestra.core.models.conditions.Condition;
+import io.kestra.core.models.flows.State;
 import io.kestra.core.models.tasks.WorkerGroup;
 import io.micronaut.core.annotation.Introspected;
 import io.swagger.v3.oas.annotations.media.Schema;
@@ -56,6 +57,12 @@ abstract public class AbstractTrigger {
 
     private Level logLevel;
 
+    @PluginProperty
+    @Schema(
+        title = "List of execution states after which a trigger should be stopped (a.k.a. disabled)."
+    )
+    private List<State.Type> stopAfter;
+
     /**
      * For backward compatibility: we rename minLogLevel to logLevel.
      * @deprecated use {@link #logLevel} instead
diff --git a/core/src/main/java/io/kestra/core/models/triggers/Trigger.java b/core/src/main/java/io/kestra/core/models/triggers/Trigger.java
index f23640730..4a121f367 100644
--- a/core/src/main/java/io/kestra/core/models/triggers/Trigger.java
+++ b/core/src/main/java/io/kestra/core/models/triggers/Trigger.java
@@ -73,6 +73,7 @@ public class Trigger extends TriggerContext {
             .flowId(flow.getId())
             .flowRevision(flow.getRevision())
             .triggerId(abstractTrigger.getId())
+            .stopAfter(abstractTrigger.getStopAfter())
             .build();
     }
 
@@ -88,6 +89,7 @@ public class Trigger extends TriggerContext {
             .triggerId(triggerContext.getTriggerId())
             .date(triggerContext.getDate())
             .backfill(triggerContext.getBackfill())
+            .stopAfter(triggerContext.getStopAfter())
             .disabled(triggerContext.getDisabled())
             .build();
     }
@@ -105,6 +107,7 @@ public class Trigger extends TriggerContext {
             .date(triggerContext.getDate())
             .nextExecutionDate(nextExecutionDate)
             .backfill(triggerContext.getBackfill())
+            .stopAfter(triggerContext.getStopAfter())
             .disabled(triggerContext.getDisabled())
             .build();
     }
@@ -126,6 +129,7 @@ public class Trigger extends TriggerContext {
             .updatedDate(Instant.now())
             .nextExecutionDate(triggerContext.getNextExecutionDate())
             .backfill(triggerContext.getBackfill())
+            .stopAfter(triggerContext.getStopAfter())
             .disabled(triggerContext.getDisabled())
             .build();
     }
@@ -148,6 +152,7 @@ public class Trigger extends TriggerContext {
             .updatedDate(Instant.now())
             .nextExecutionDate(nextExecutionDate)
             .backfill(triggerContext.getBackfill())
+            .stopAfter(triggerContext.getStopAfter())
             .disabled(triggerContext.getDisabled())
             .build();
     }
@@ -170,6 +175,7 @@ public class Trigger extends TriggerContext {
             .executionCurrentState(execution.getState().getCurrent())
             .updatedDate(Instant.now())
             .backfill(trigger.getBackfill())
+            .stopAfter(trigger.getStopAfter())
             .disabled(trigger.getDisabled())
             .build();
     }
@@ -191,6 +197,7 @@ public class Trigger extends TriggerContext {
             .evaluateRunningDate(evaluateRunningDate)
             .updatedDate(Instant.now())
             .backfill(trigger.getBackfill())
+            .stopAfter(trigger.getStopAfter())
             .disabled(trigger.getDisabled())
             .build();
     }
@@ -204,6 +211,7 @@ public class Trigger extends TriggerContext {
             .triggerId(abstractTrigger.getId())
             .date(ZonedDateTime.now().truncatedTo(ChronoUnit.SECONDS))
             .nextExecutionDate(((PollingTriggerInterface) abstractTrigger).nextEvaluationDate(conditionContext, lastTrigger))
+            .stopAfter(abstractTrigger.getStopAfter())
             .disabled(lastTrigger.map(TriggerContext::getDisabled).orElse(Boolean.FALSE))
             .build();
     }
@@ -234,7 +242,24 @@ public class Trigger extends TriggerContext {
             .build();
     }
 
-    public Trigger resetExecution() {
+    public Trigger resetExecution(State.Type executionEndState) {
+        // switch disabled automatically if the executionEndState is one of the stopAfter states
+        Boolean disabled = this.getStopAfter() != null ? this.getStopAfter().contains(executionEndState) : this.getDisabled();
+
+        return Trigger.builder()
+            .tenantId(this.getTenantId())
+            .namespace(this.getNamespace())
+            .flowId(this.getFlowId())
+            .flowRevision(this.getFlowRevision())
+            .triggerId(this.getTriggerId())
+            .date(this.getDate())
+            .nextExecutionDate(this.getNextExecutionDate())
+            .stopAfter(this.getStopAfter())
+            .disabled(disabled)
+            .build();
+    }
+
+    public Trigger unlock() {
         return Trigger.builder()
             .tenantId(this.getTenantId())
             .namespace(this.getNamespace())
@@ -244,8 +269,11 @@ public class Trigger extends TriggerContext {
             .date(this.getDate())
             .nextExecutionDate(this.getNextExecutionDate())
             .backfill(this.getBackfill())
+            .stopAfter(this.getStopAfter())
+            .disabled(this.getDisabled())
             .build();
     }
+
     // if the next date is after the backfill end, we remove the backfill
     // if not, we update the backfill with the next Date
     // which will be the base date to calculate the next one
diff --git a/core/src/main/java/io/kestra/core/models/triggers/TriggerContext.java b/core/src/main/java/io/kestra/core/models/triggers/TriggerContext.java
index 07ad964b9..066186c2e 100644
--- a/core/src/main/java/io/kestra/core/models/triggers/TriggerContext.java
+++ b/core/src/main/java/io/kestra/core/models/triggers/TriggerContext.java
@@ -1,5 +1,6 @@
 package io.kestra.core.models.triggers;
 
+import io.kestra.core.models.flows.State;
 import io.kestra.core.utils.IdUtils;
 import io.micronaut.core.annotation.Introspected;
 import io.micronaut.core.annotation.Nullable;
@@ -12,6 +13,7 @@ import lombok.ToString;
 import lombok.experimental.SuperBuilder;
 
 import java.time.ZonedDateTime;
+import java.util.List;
 
 @SuperBuilder(toBuilder = true)
 @ToString
@@ -43,6 +45,9 @@ public class TriggerContext {
     @Nullable
     private Backfill backfill;
 
+    @Nullable
+    private List<State.Type> stopAfter;
+
     @Builder.Default
     private Boolean disabled = Boolean.FALSE;
 
diff --git a/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java b/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
index f12746d80..2be39f0c6 100644
--- a/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
+++ b/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
@@ -199,6 +199,7 @@ public abstract class AbstractScheduler implements Scheduler {
                             .triggerId(abstractTrigger.getId())
                             .date(now())
                             .nextExecutionDate(nextExecutionDate)
+                            .stopAfter(abstractTrigger.getStopAfter())
                             .build();
                         this.triggerState.create(newTrigger);
                     } catch (Exception e) {
@@ -274,8 +275,11 @@ public abstract class AbstractScheduler implements Scheduler {
             }
 
             triggers.forEach(trigger -> schedulableNextDate.remove(trigger.uid()));
+            List<Trigger> triggerContextsToEvaluate = triggers.stream()
+                .filter(trigger -> Boolean.FALSE.equals(trigger.getDisabled()))
+                .toList();
 
-            List<FlowWithTriggers> schedulable = this.computeSchedulable(flowListeners.flows(), triggers, scheduleContext);
+            List<FlowWithTriggers> schedulable = this.computeSchedulable(flowListeners.flows(), triggerContextsToEvaluate, scheduleContext);
 
             metricRegistry
                 .counter(MetricRegistry.SCHEDULER_LOOP_COUNT)
@@ -299,7 +303,7 @@ public abstract class AbstractScheduler implements Scheduler {
                     .abstractTrigger(flowWithTriggers.getAbstractTrigger())
                     .pollingTrigger((PollingTriggerInterface) flowWithTriggers.getAbstractTrigger())
                     .conditionContext(flowWithTriggers.getConditionContext())
-                    .triggerContext(flowWithTriggers.TriggerContext.toBuilder().date(now()).build())
+                    .triggerContext(flowWithTriggers.TriggerContext.toBuilder().date(now()).stopAfter(flowWithTriggers.getAbstractTrigger().getStopAfter()).build())
                     .build())
                 .filter(f -> f.getTriggerContext().getEvaluateRunningDate() == null && !f.getTriggerContext().getDisabled())
                 .filter(this::isExecutionNotRunning)
@@ -393,6 +397,7 @@ public abstract class AbstractScheduler implements Scheduler {
                         executionWithTrigger.getTriggerContext(),
                         executionWithTrigger.getExecution()
                     );
+
                     // Check if the localTriggerState contains it
                     // however, its mean it has been deleted during the execution time
                     this.triggerState.update(trigger);
@@ -670,6 +675,7 @@ public abstract class AbstractScheduler implements Scheduler {
                     .date(f.getTriggerContext().getNextExecutionDate())
                     .nextExecutionDate(f.getTriggerContext().getNextExecutionDate())
                     .backfill(f.getTriggerContext().getBackfill())
+                    .stopAfter(f.getTriggerContext().getStopAfter())
                     .build()
                 )
                 .next(f.getTriggerContext().getNextExecutionDate())
diff --git a/core/src/main/java/io/kestra/core/schedulers/DefaultScheduler.java b/core/src/main/java/io/kestra/core/schedulers/DefaultScheduler.java
index 11257313e..fe738c97b 100644
--- a/core/src/main/java/io/kestra/core/schedulers/DefaultScheduler.java
+++ b/core/src/main/java/io/kestra/core/schedulers/DefaultScheduler.java
@@ -63,7 +63,7 @@ public class DefaultScheduler extends AbstractScheduler {
                 Trigger trigger = Await.until(()  -> watchingTrigger.get(execution.getId()), Duration.ofSeconds(5));
                 var flow = flowRepository.findById(execution.getTenantId(), execution.getNamespace(), execution.getFlowId()).orElse(null);
                 if (execution.isDeleted() || conditionService.isTerminatedWithListeners(flow, execution)) {
-                    triggerState.update(trigger.resetExecution());
+                    triggerState.update(trigger.resetExecution(execution.getState().getCurrent()));
                     watchingTrigger.remove(execution.getId());
                 } else {
                     triggerState.update(Trigger.of(execution, trigger));
diff --git a/core/src/main/java/io/kestra/core/utils/Hashing.java b/core/src/main/java/io/kestra/core/utils/Hashing.java
index 6ce7b1da2..c0164403a 100644
--- a/core/src/main/java/io/kestra/core/utils/Hashing.java
+++ b/core/src/main/java/io/kestra/core/utils/Hashing.java
@@ -2,6 +2,10 @@ package io.kestra.core.utils;
 
 import com.google.common.base.Charsets;
 import com.google.common.hash.HashCode;
+import com.google.common.hash.Hasher;
+
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
 
 /**
  * Utilities for hashing.
@@ -30,6 +34,33 @@ public final class Hashing {
         return getHashString(value).asLong();
     }
 
+    /**
+     * Hashes the given value using SHA-512 algorithm.
+     *
+     * @param value     the value to be hashed.
+     * @param salt      an optional salt to be added to the value.
+     * @return          the digest.
+     */
+    public static byte[] sha512Hash(final byte[] value, byte[] salt) {
+        try {
+            MessageDigest md = MessageDigest.getInstance("SHA-512");
+            if (salt != null && salt.length > 0) {
+                md.update(salt);
+            }
+            return md.digest(value);
+        } catch (NoSuchAlgorithmException e) {
+            throw new AssertionError(e);
+        }
+    }
+
+    public static String encodeBytesToHex(final byte[] bytes) {
+        return HashCode.fromBytes(bytes).toString();
+    }
+
+    public static byte[] decodeHexToBytes(final String value) {
+        return HashCode.fromString(value).asBytes();
+    }
+
     private static HashCode getHashString(String value) {
         return com.google.common.hash.Hashing.murmur3_128().hashString(value, Charsets.UTF_8);
     }
diff --git a/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcTriggerRepository.java b/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcTriggerRepository.java
index 94bf15380..ed5d4e6b9 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcTriggerRepository.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcTriggerRepository.java
@@ -181,30 +181,6 @@ public abstract class AbstractJdbcTriggerRepository extends AbstractJdbcReposito
             });
     }
 
-    // update/reset execution need to be done in a transaction
-    // to be sure we get the correct date/nextDate when updating
-    public Trigger resetExecution(Trigger trigger) {
-        return this.jdbcRepository
-            .getDslContextWrapper()
-            .transactionResult(configuration -> {
-                DSLContext context = DSL.using(configuration);
-                Optional<Trigger> optionalTrigger = this.jdbcRepository.fetchOne(context.select(field("value"))
-                    .from(this.jdbcRepository.getTable())
-                    .where(
-                        field("key").eq(trigger.uid())
-                    ).forUpdate());
-
-                if (optionalTrigger.isPresent()) {
-                    Trigger current = optionalTrigger.get();
-                    this.save(context, current.resetExecution());
-
-                    return current;
-                }
-
-                return null;
-            });
-    }
-
     // Allow to update a trigger from a flow & an abstract trigger
     // using forUpdate to avoid the lastTrigger to be updated by another thread
     // before doing the update
diff --git a/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcWorkerInstanceRepository.java b/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcWorkerInstanceRepository.java
index 515595688..75c63aa49 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcWorkerInstanceRepository.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcWorkerInstanceRepository.java
@@ -74,7 +74,7 @@ public abstract class AbstractJdbcWorkerInstanceRepository extends AbstractJdbcR
         workerInstance.ifPresent(heartbeat -> {
             heartbeat.setStatus(WorkerInstance.Status.DEAD);
 
-            log.warn("Detected evicted worker: {}", heartbeat);
+            log.warn("Detected non-responding worker, stated to DEAD: {}", heartbeat);
 
             this.jdbcRepository.persist(heartbeat, context,  this.jdbcRepository.persistFields(heartbeat));
         });
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
index bb2e9dc1c..4f7c047de 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
@@ -4,11 +4,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import io.kestra.core.exceptions.DeserializationException;
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.metrics.MetricRegistry;
-import io.kestra.core.models.executions.Execution;
-import io.kestra.core.models.executions.ExecutionKilled;
-import io.kestra.core.models.executions.LogEntry;
-import io.kestra.core.models.executions.TaskRun;
-import io.kestra.core.models.executions.TaskRunAttempt;
+import io.kestra.core.models.executions.*;
 import io.kestra.core.models.executions.statistics.ExecutionCount;
 import io.kestra.core.models.flows.Concurrency;
 import io.kestra.core.models.flows.Flow;
@@ -21,10 +17,7 @@ import io.kestra.core.models.triggers.multipleflows.MultipleConditionStorageInte
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
 import io.kestra.core.repositories.FlowRepositoryInterface;
-import io.kestra.core.runners.DefaultFlowExecutor;
-import io.kestra.core.runners.ExecutableUtils;
 import io.kestra.core.runners.Executor;
-import io.kestra.core.runners.ExecutorInterface;
 import io.kestra.core.runners.ExecutorService;
 import io.kestra.core.runners.*;
 import io.kestra.core.services.*;
@@ -48,17 +41,12 @@ import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.tuple.Pair;
 import org.slf4j.event.Level;
-import reactor.core.publisher.Flux;
 
 import java.io.IOException;
 import java.time.Duration;
 import java.util.List;
 import java.util.Optional;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -340,6 +328,7 @@ public class JdbcExecutor implements ExecutorInterface {
                 });
 
             workersToDelete.forEach(worker -> {
+                log.warn("Deleted dead worker: {}", worker);
                 workerInstanceRepository.delete(context, worker);
             });
 
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcHeartbeat.java b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcHeartbeat.java
index a508408ab..1f9982146 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcHeartbeat.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcHeartbeat.java
@@ -82,9 +82,7 @@ public class JdbcHeartbeat {
 
             if (workerInstanceRepository.heartbeatCheckUp(workerInstance.getWorkerUuid().toString()).isEmpty()) {
                 log.error("heartbeatCheckUp failed, unable to find current instance '{}', Shutting down now!", workerInstance.getWorkerUuid());
-                if (serverType == ServerType.WORKER) {
-                    Runtime.getRuntime().exit(1);
-                }
+                Runtime.getRuntime().exit(1);
             }
         }
     }
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcScheduler.java b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcScheduler.java
index 4426552fd..10ca77e1f 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcScheduler.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcScheduler.java
@@ -75,7 +75,7 @@ public class JdbcScheduler extends AbstractScheduler {
                         triggerRepository
                             .findByExecution(execution)
                             .ifPresent(trigger -> {
-                                ((JdbcSchedulerTriggerState) this.triggerState).resetExecution(trigger.resetExecution());
+                                this.triggerState.update(trigger.resetExecution(execution.getState().getCurrent()));
                             });
                     } else {
                         // update execution state on each state change so the scheduler knows the execution is running
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcSchedulerTriggerState.java b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcSchedulerTriggerState.java
index 339814d24..8f47587c6 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcSchedulerTriggerState.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcSchedulerTriggerState.java
@@ -71,11 +71,6 @@ public class JdbcSchedulerTriggerState implements SchedulerTriggerStateInterface
         return this.triggerRepository.updateExecution(trigger);
     }
 
-    public Trigger resetExecution(Trigger trigger) {
-
-        return this.triggerRepository.resetExecution(trigger);
-    }
-
     public Trigger update(Flow flow, AbstractTrigger abstractTrigger, ConditionContext conditionContext) {
         return this.triggerRepository.update(flow, abstractTrigger, conditionContext);
     }
diff --git a/webserver/src/main/java/io/kestra/webserver/controllers/TriggerController.java b/webserver/src/main/java/io/kestra/webserver/controllers/TriggerController.java
index 6cab96b95..406a58df3 100644
--- a/webserver/src/main/java/io/kestra/webserver/controllers/TriggerController.java
+++ b/webserver/src/main/java/io/kestra/webserver/controllers/TriggerController.java
@@ -75,7 +75,7 @@ public class TriggerController {
             throw new IllegalStateException("Trigger is not locked");
         }
 
-        trigger = trigger.resetExecution();
+        trigger = trigger.unlock();
         triggerQueue.emit(trigger);
 
         return HttpResponse.ok(trigger);


Test patch:
diff --git a/core/src/test/java/io/kestra/core/docs/JsonSchemaGeneratorTest.java b/core/src/test/java/io/kestra/core/docs/JsonSchemaGeneratorTest.java
index 4995d13ef..fed1a8ca2 100644
--- a/core/src/test/java/io/kestra/core/docs/JsonSchemaGeneratorTest.java
+++ b/core/src/test/java/io/kestra/core/docs/JsonSchemaGeneratorTest.java
@@ -123,9 +123,11 @@ class JsonSchemaGeneratorTest {
 
             Map<String, Object> jsonSchema = jsonSchemaGenerator.generate(AbstractTrigger.class, AbstractTrigger.class);
 
+            System.out.println(jsonSchema.get("properties"));
             assertThat((Map<String, Object>) jsonSchema.get("properties"), allOf(
-                Matchers.aMapWithSize(1),
-                hasKey("conditions")
+                Matchers.aMapWithSize(2),
+                hasKey("conditions"),
+                hasKey("stopAfter")
             ));
         });
     }
diff --git a/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java b/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java
index c753da852..fdf84b1b2 100644
--- a/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java
@@ -29,7 +29,7 @@ import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 
-@MicronautTest(transactional = false)
+@MicronautTest(transactional = false, rebuildContext = true) // rebuild context to lower possible flaky tests
 abstract public class AbstractSchedulerTest {
     @Inject
     protected ApplicationContext applicationContext;
diff --git a/core/src/test/java/io/kestra/core/schedulers/SchedulerPollingTriggerTest.java b/core/src/test/java/io/kestra/core/schedulers/SchedulerPollingTriggerTest.java
index 6db779184..17e518d40 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerPollingTriggerTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerPollingTriggerTest.java
@@ -1,30 +1,34 @@
 package io.kestra.core.schedulers;
 
 import io.kestra.core.models.executions.Execution;
-import io.kestra.core.queues.QueueFactoryInterface;
-import io.kestra.core.queues.QueueInterface;
-import io.kestra.core.repositories.LocalFlowRepositoryLoader;
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.flows.State;
+import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.runners.FlowListeners;
+import io.kestra.core.runners.TestMethodScopedWorker;
 import io.kestra.core.runners.Worker;
+import io.kestra.core.tasks.executions.Fail;
+import io.kestra.core.tasks.test.PollingTrigger;
+import io.kestra.core.utils.Await;
 import io.micronaut.context.ApplicationContext;
 import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
 import jakarta.inject.Inject;
-import jakarta.inject.Named;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
-import java.util.Objects;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.*;
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.spy;
 
-@Disabled("Causes all sort of issues on CI")
-@MicronautTest(transactional = false)
-public class SchedulerPollingTriggerTest {
+@MicronautTest(transactional = false, rebuildContext = true) // without 'rebuildContext = true' the second test fail
+public class SchedulerPollingTriggerTest extends AbstractSchedulerTest {
     @Inject
     private ApplicationContext applicationContext;
 
@@ -34,39 +38,78 @@ public class SchedulerPollingTriggerTest {
     @Inject
     private FlowListeners flowListenersService;
 
-    @Inject
-    @Named(QueueFactoryInterface.EXECUTION_NAMED)
-    private QueueInterface<Execution> executionQueue;
-
-    @Inject
-    private LocalFlowRepositoryLoader repositoryLoader;
-
 
     @Test
     void pollingTrigger() throws Exception {
+        // mock flow listener
+        FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
+        PollingTrigger pollingTrigger = createPollingTrigger(null);
+        Flow flow = createPollingTriggerFlow(pollingTrigger);
+        doReturn(List.of(flow))
+            .when(flowListenersServiceSpy)
+            .flows();
+
         CountDownLatch queueCount = new CountDownLatch(1);
 
         try (
-            AbstractScheduler scheduler = new DefaultScheduler(
-                this.applicationContext,
-                this.flowListenersService,
-                this.triggerState
-            );
-            Worker worker = new Worker(applicationContext, 8, null)
+            AbstractScheduler scheduler = scheduler(flowListenersServiceSpy);
+            Worker worker = new TestMethodScopedWorker(applicationContext, 8, null)
         ) {
             AtomicReference<Execution> last = new AtomicReference<>();
 
             Runnable executionQueueStop = executionQueue.receive(execution -> {
-                last.set(execution.getLeft());
-
-                queueCount.countDown();
-                assertThat(execution.getLeft().getFlowId(), is("polling-trigger"));
+                if (execution.getLeft().getFlowId().equals(flow.getId())) {
+                    last.set(execution.getLeft());
+                    queueCount.countDown();
+                }
             });
 
             worker.run();
             scheduler.run();
 
-            repositoryLoader.load(Objects.requireNonNull(SchedulerPollingTriggerTest.class.getClassLoader().getResource("flows/trigger")));
+            queueCount.await(10, TimeUnit.SECONDS);
+            // close the execution queue consumer
+            executionQueueStop.run();
+
+            assertThat(queueCount.getCount(), is(0L));
+            assertThat(last.get(), notNullValue());
+        }
+    }
+
+    @Test
+    void pollingTriggerStopAfter() throws Exception {
+        // mock flow listener
+        FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
+        PollingTrigger pollingTrigger = createPollingTrigger(List.of(State.Type.FAILED));
+        Flow flow = createPollingTriggerFlow(pollingTrigger)
+            .toBuilder()
+            .tasks(List.of(Fail.builder().id("fail").type(Fail.class.getName()).build()))
+            .build();
+        doReturn(List.of(flow))
+            .when(flowListenersServiceSpy)
+            .flows();
+
+        CountDownLatch queueCount = new CountDownLatch(2);
+
+        try (
+            AbstractScheduler scheduler = scheduler(flowListenersServiceSpy);
+            Worker worker = new TestMethodScopedWorker(applicationContext, 8, null)
+        ) {
+            AtomicReference<Execution> last = new AtomicReference<>();
+
+            Runnable executionQueueStop = executionQueue.receive(execution -> {
+                if (execution.getLeft().getFlowId().equals(flow.getId())) {
+                    last.set(execution.getLeft());
+                    queueCount.countDown();
+
+                    if (execution.getLeft().getState().getCurrent() == State.Type.CREATED) {
+                        executionQueue.emit(execution.getLeft().withState(State.Type.FAILED));
+                    }
+                }
+            });
+
+            worker.run();
+            scheduler.run();
 
             queueCount.await(10, TimeUnit.SECONDS);
             // close the execution queue consumer
@@ -74,6 +117,32 @@ public class SchedulerPollingTriggerTest {
 
             assertThat(queueCount.getCount(), is(0L));
             assertThat(last.get(), notNullValue());
+
+            // Assert that the trigger is now disabled.
+            // It needs to await on assertion as it will be disabled AFTER we receive a success execution.
+            Trigger trigger = Trigger.of(flow, pollingTrigger);
+            Await.until(() -> this.triggerState.findLast(trigger).map(t -> t.getDisabled()).orElse(false), Duration.ofMillis(100), Duration.ofSeconds(10));
         }
     }
+
+    private Flow createPollingTriggerFlow(PollingTrigger pollingTrigger) {
+        return createFlow(Collections.singletonList(pollingTrigger));
+    }
+
+    private PollingTrigger createPollingTrigger(List<State.Type> stopAfter) {
+        return PollingTrigger.builder()
+            .id("polling-trigger")
+            .type(PollingTrigger.class.getName())
+            .duration(500L)
+            .stopAfter(stopAfter)
+            .build();
+    }
+
+    private AbstractScheduler scheduler(FlowListeners flowListenersServiceSpy) {
+        return new DefaultScheduler(
+            applicationContext,
+            flowListenersServiceSpy,
+            triggerState
+        );
+    }
 }
diff --git a/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleTest.java b/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleTest.java
index 3f8eb81e3..7555f0259 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleTest.java
@@ -40,21 +40,24 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
     @Named(QueueFactoryInterface.WORKERTASKLOG_NAMED)
     protected QueueInterface<LogEntry> logQueue;
 
-    private static Flow createScheduleFlow(String zone, String triggerId, boolean invalid) {
-        Schedule schedule = Schedule.builder()
+    private Schedule.ScheduleBuilder<?, ?> createScheduleTrigger(String zone, String cron, String triggerId, boolean invalid) {
+        return Schedule.builder()
             .id(triggerId + (invalid ? "-invalid" : ""))
             .type(Schedule.class.getName())
-            .cron("0 * * * *")
+            .cron(cron)
             .timezone(zone)
             .inputs(Map.of(
                 "testInputs", "test-inputs"
-            ))
-            .build();
+            ));
+    }
+
+    private Flow createScheduleFlow(String zone, String triggerId, boolean invalid) {
+        Schedule schedule = createScheduleTrigger(zone, "0 * * * *", triggerId, invalid).build();
 
         return createFlow(Collections.singletonList(schedule));
     }
 
-    private static ZonedDateTime date(int minus) {
+    private ZonedDateTime date(int minus) {
         return ZonedDateTime.now()
             .minusHours(minus)
             .truncatedTo(ChronoUnit.HOURS);
@@ -285,4 +288,58 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
             assertThat(lastTrigger.getNextExecutionDate().truncatedTo(ChronoUnit.HOURS).isEqual(now), is(true));
         }
     }
+
+    @Test
+    void stopAfterSchedule() throws Exception {
+        // mock flow listeners
+        FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
+        Schedule schedule = createScheduleTrigger("Europe/Paris", "* * * * *", "stopAfter", false).build();
+        Flow flow = createFlow(Collections.singletonList(schedule));
+        doReturn(List.of(flow))
+            .when(flowListenersServiceSpy)
+            .flows();
+
+        // to avoid waiting too much before a trigger execution, we add a last trigger with a date now - 1m.
+        Trigger lastTrigger = Trigger
+            .builder()
+            .triggerId("stopAfter")
+            .flowId(flow.getId())
+            .namespace(flow.getNamespace())
+            .date(ZonedDateTime.now().minusMinutes(1L))
+            .build();
+        triggerState.create(lastTrigger);
+
+        CountDownLatch queueCount = new CountDownLatch(2);
+
+        // scheduler
+        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy);
+             Worker worker = new TestMethodScopedWorker(applicationContext, 8, null)) {
+            // wait for execution
+            Runnable assertionStop = executionQueue.receive(either -> {
+                Execution execution = either.getLeft();
+                assertThat(execution.getInputs().get("testInputs"), is("test-inputs"));
+                assertThat(execution.getInputs().get("def"), is("awesome"));
+                assertThat(execution.getFlowId(), is(flow.getId()));
+
+                queueCount.countDown();
+                if (execution.getState().getCurrent() == State.Type.CREATED) {
+                    executionQueue.emit(execution.withState(State.Type.SUCCESS));
+                }
+            });
+
+            worker.run();
+            scheduler.run();
+
+            queueCount.await(1, TimeUnit.MINUTES);
+            // needed for RetryingTest to work since there is no context cleaning between method => we have to clear assertion receiver manually
+            assertionStop.run();
+
+            assertThat(queueCount.getCount(), is(0L));
+
+            // Assert that the trigger is now disabled.
+            // It needs to await on assertion as it will be disabled AFTER we receive a success execution.
+            Trigger trigger = Trigger.of(flow, schedule);
+            Await.until(() -> this.triggerState.findLast(trigger).map(t -> t.getDisabled()).orElse(false), Duration.ofMillis(100), Duration.ofSeconds(10));
+        }
+    }
 }
diff --git a/core/src/test/resources/flows/trigger/polling-trigger.yaml b/core/src/test/resources/flows/trigger/polling-trigger.yaml
deleted file mode 100644
index 4f5e4b98b..000000000
--- a/core/src/test/resources/flows/trigger/polling-trigger.yaml
+++ /dev/null
@@ -1,11 +0,0 @@
-id: polling-trigger
-namespace: io.kestra.tests.trigger
-
-triggers:
-  - id: watch
-    type: io.kestra.core.tasks.test.PollingTrigger
-
-tasks:
-  - id: end
-    type: io.kestra.core.tasks.debugs.Return
-    format: "{{task.id}} > {{taskrun.startDate}}"