Repository: kestra-io/kestra (java)

Issue:
Allow disabling system backfills on crash
1. deprecate `backfill` property in the Schedule trigger source code
2. add `recoverMissedSchedules` Enum property set by default to ALL (by default backfilling all missed scheduled executions)
3. add a global configuration on the instance as shown below (best if we use the same naming as on the property for easy understanding) -- this configuration is taken into account if not explicitly set in the trigger definition
    
syntax for the Schedule property:

```yaml
triggers:
  - id: schedule
    type: io.kestra.core.models.triggers.types.Schedule
    cron: "*/15 * * * *"
    recoverMissedSchedules: LAST | NONE | ALL # ⚡️ new! 
    backfill: # ❌ show deprecation when this property is used 
```

syntax for the global configuration:
```yaml
kestra:
  server:
    recover-missed-schedules: ALL # can be also set globally to NONE or LAST
```

Pull request description:
feat(core): Recover Missed Schedule
Fixes #2219

Allow to specify the behavior in case of missed schedules: recover all, recover last or recover none.


Patch:
diff --git a/cli/src/main/resources/application.yml b/cli/src/main/resources/application.yml
index 4bd0ebd6c..700e4bd5a 100644
--- a/cli/src/main/resources/application.yml
+++ b/cli/src/main/resources/application.yml
@@ -187,6 +187,9 @@ kestra:
         values:
           outputs:
             enabled: true # backward-compatibility with version prior to v0.15.0
+      - type: io.kestra.core.models.triggers.types.Schedule
+        values:
+          recover-missed-schedules: ALL
   variables:
     env-vars-prefix: KESTRA_
     cache-enabled: true
diff --git a/core/src/main/java/io/kestra/core/models/triggers/types/Schedule.java b/core/src/main/java/io/kestra/core/models/triggers/types/Schedule.java
index 1b521adb3..e2f19639a 100644
--- a/core/src/main/java/io/kestra/core/models/triggers/types/Schedule.java
+++ b/core/src/main/java/io/kestra/core/models/triggers/types/Schedule.java
@@ -117,6 +117,8 @@ import java.util.*;
 
 )
 public class Schedule extends AbstractTrigger implements PollingTriggerInterface, TriggerOutput<Schedule.Output> {
+    private static final String PLUGIN_PROPERTY_RECOVER_MISSED_SCHEDULES = "recover-missed-schedules";
+
     public static final CronParser CRON_PARSER = new CronParser(CronDefinitionBuilder.defineCron()
         .withMinutes().withValidRange(0, 59).withStrictRange().and()
         .withHours().withValidRange(0, 23).withStrictRange().and()
@@ -194,6 +196,14 @@ public class Schedule extends AbstractTrigger implements PollingTriggerInterface
     @Deprecated
     private ScheduleBackfill backfill;
 
+    @Schema(
+        title = "What to do in case of missed schedules",
+        description = "`ALL` will recover all missed schedules, `LAST`  will only recovered the last missing one, `NONE` will not recover any missing schedule.\n" +
+            "The default is `ALL` unless a different value is configured using the global plugin configuration."
+    )
+    @PluginProperty
+    private RecoverMissedSchedules recoverMissedSchedules;
+
     @Override
     public ZonedDateTime nextEvaluationDate(ConditionContext conditionContext, Optional<? extends TriggerContext> last) {
         ExecutionTime executionTime = this.executionTime();
@@ -203,10 +213,11 @@ public class Schedule extends AbstractTrigger implements PollingTriggerInterface
             ZonedDateTime lastDate;
             if (last.get().getBackfill() != null) {
                 backfill = last.get().getBackfill();
-                lastDate = backfill.getCurrentDate();
+                lastDate = convertDateTime(backfill.getCurrentDate());
             } else {
                 lastDate = convertDateTime(last.get().getDate());
             }
+
             // previous present & scheduleConditions
             if (this.scheduleConditions != null) {
                 try {
@@ -216,14 +227,15 @@ public class Schedule extends AbstractTrigger implements PollingTriggerInterface
                         lastDate,
                         true
                     );
+
                     if (next.isPresent()) {
                         return next.get().truncatedTo(ChronoUnit.SECONDS);
                     }
                 } catch (InternalException e) {
                     conditionContext.getRunContext().logger().warn("Unable to evaluate the conditions for the next evaluation date for trigger '{}', conditions will not be evaluated", this.getId());
                 }
-
             }
+
             // previous present but no scheduleConditions
             nextDate = computeNextEvaluationDate(executionTime, lastDate).orElse(null);
 
@@ -234,12 +246,12 @@ public class Schedule extends AbstractTrigger implements PollingTriggerInterface
                 nextDate = computeNextEvaluationDate(executionTime, ZonedDateTime.now()).orElse(null);
             }
         }
-        // no previous present & no backfill, just provide now
+        // no previous present & no backfill or recover missed schedules, just provide now
         else {
             nextDate = computeNextEvaluationDate(executionTime, ZonedDateTime.now()).orElse(null);
         }
-        // if max delay reach, we calculate a new date
-        // except if we are doing a backfill
+
+        // if max delay reached, we calculate a new date except if we are doing a backfill
         if (this.lateMaximumDelay != null && nextDate != null && backfill == null) {
             Output scheduleDates = this.scheduleDates(executionTime, nextDate).orElse(null);
             scheduleDates = this.handleMaxDelay(scheduleDates);
@@ -253,6 +265,34 @@ public class Schedule extends AbstractTrigger implements PollingTriggerInterface
         return nextDate;
     }
 
+    @Override
+    public ZonedDateTime nextEvaluationDate() {
+        // it didn't take into account the schedule condition, but as they are taken into account inside eval() it's OK.
+        ExecutionTime executionTime = this.executionTime();
+        return computeNextEvaluationDate(executionTime, ZonedDateTime.now()).orElse(ZonedDateTime.now());
+    }
+
+    public ZonedDateTime previousEvaluationDate(ConditionContext conditionContext) {
+        ExecutionTime executionTime = this.executionTime();
+        if (this.scheduleConditions != null) {
+            try {
+                Optional<ZonedDateTime> previous = this.truePreviousNextDateWithCondition(
+                    executionTime,
+                    conditionContext,
+                    ZonedDateTime.now(),
+                    false
+                );
+
+                if (previous.isPresent()) {
+                    return previous.get().truncatedTo(ChronoUnit.SECONDS);
+                }
+            } catch (InternalException e) {
+                conditionContext.getRunContext().logger().warn("Unable to evaluate the conditions for the next evaluation date for trigger '{}', conditions will not be evaluated", this.getId());
+            }
+        }
+        return computePreviousEvaluationDate(executionTime, ZonedDateTime.now()).orElse(ZonedDateTime.now());
+    }
+
     @Override
     public Optional<Execution> evaluate(ConditionContext conditionContext, TriggerContext triggerContext) throws Exception {
         RunContext runContext = conditionContext.getRunContext();
@@ -368,7 +408,14 @@ public class Schedule extends AbstractTrigger implements PollingTriggerInterface
         return Optional.of(execution);
     }
 
-    private static List<Label> getLabels(ConditionContext conditionContext, Backfill backfill) {
+    public RecoverMissedSchedules defaultRecoverMissedSchedules(RunContext runContext) {
+        return runContext
+            .<String>pluginConfiguration(PLUGIN_PROPERTY_RECOVER_MISSED_SCHEDULES)
+            .map(conf -> RecoverMissedSchedules.valueOf(conf))
+            .orElse(RecoverMissedSchedules.ALL);
+    }
+
+    private List<Label> getLabels(ConditionContext conditionContext, Backfill backfill) {
         List<Label> labels = conditionContext.getFlow().getLabels() != null ? conditionContext.getFlow().getLabels() : new ArrayList<>();
         if (backfill != null && backfill.getLabels() != null) {
             labels.addAll(backfill.getLabels());
@@ -428,6 +475,10 @@ public class Schedule extends AbstractTrigger implements PollingTriggerInterface
         return executionTime.nextExecution(date).map(zonedDateTime -> zonedDateTime.truncatedTo(ChronoUnit.SECONDS));
     }
 
+    private Optional<ZonedDateTime> computePreviousEvaluationDate(ExecutionTime executionTime, ZonedDateTime date) {
+        return executionTime.lastExecution(date).map(zonedDateTime -> zonedDateTime.truncatedTo(ChronoUnit.SECONDS));
+    }
+
     private Output trueOutputWithCondition(ExecutionTime executionTime, ConditionContext conditionContext, Output output) throws InternalException {
         Output.OutputBuilder<?, ?> outputBuilder = Output.builder()
             .date(output.getDate());
@@ -535,5 +586,12 @@ public class Schedule extends AbstractTrigger implements PollingTriggerInterface
         @Schema(
             title = "The first start date."
         )
-        ZonedDateTime start;}
+        ZonedDateTime start;
+    }
+
+    public enum RecoverMissedSchedules {
+        LAST,
+        NONE,
+        ALL
+    }
 }
diff --git a/core/src/main/java/io/kestra/core/runners/RunContext.java b/core/src/main/java/io/kestra/core/runners/RunContext.java
index 758e52eba..9d8475936 100644
--- a/core/src/main/java/io/kestra/core/runners/RunContext.java
+++ b/core/src/main/java/io/kestra/core/runners/RunContext.java
@@ -458,6 +458,7 @@ public class RunContext {
             context,
             storageInterface
         );
+        this.initPluginConfiguration(applicationContext, trigger.getType());
         return this;
     }
 
@@ -491,6 +492,7 @@ public class RunContext {
 
         this.variables = ImmutableMap.copyOf(clone);
         this.storage = new InternalStorage(logger(), StorageContext.forTask(taskRun), storageInterface);
+        this.initPluginConfiguration(applicationContext, workerTask.getTask().getType());
         return this;
     }
 
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 0baff5138..b6b1dc0f8 100644
--- a/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
+++ b/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
@@ -12,6 +12,7 @@ import io.kestra.core.models.flows.State;
 import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.models.triggers.PollingTriggerInterface;
 import io.kestra.core.models.triggers.Trigger;
+import io.kestra.core.models.triggers.TriggerContext;
 import io.kestra.core.models.triggers.types.Schedule;
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
@@ -60,7 +61,9 @@ public abstract class AbstractScheduler implements Scheduler {
     private final TaskDefaultService taskDefaultService;
     private final WorkerGroupService workerGroupService;
     private final LogService logService;
-    protected Boolean isReady = false;
+
+    // must be volatile as it's updated by the flow listener thread and read by the scheduleExecutor thread
+    private volatile Boolean isReady = false;
 
     private final ScheduledExecutorService scheduleExecutor = Executors.newSingleThreadScheduledExecutor();
 
@@ -92,6 +95,10 @@ public abstract class AbstractScheduler implements Scheduler {
         this.logService = applicationContext.getBean(LogService.class);
     }
 
+    protected boolean isReady() {
+        return isReady;
+    }
+
     @Override
     public void run() {
         this.flowListeners.run();
@@ -176,37 +183,58 @@ public abstract class AbstractScheduler implements Scheduler {
         );
     }
 
-    // Initialized local trigger state
-    // and if some flows were created outside the box, for example from CLI
-    // then we may have some triggers that are not created yet
+    // Initialized local trigger state,
+    // and if some flows were created outside the box, for example from the CLI,
+    // then we may have some triggers that are not created yet.
     private void initializedTriggers(List<Flow> flows) {
+        record FlowAndTrigger(Flow flow, AbstractTrigger trigger) {}
         List<Trigger> triggers = triggerState.findAllForAllTenants();
 
-        flows.forEach(flow -> {
-            ListUtils.emptyOnNull(flow.getTriggers()).forEach(abstractTrigger -> {
-                if (triggers.stream().noneMatch(trigger -> trigger.uid().equals(Trigger.uid(flow, abstractTrigger))) && abstractTrigger instanceof PollingTriggerInterface pollingAbstractTrigger) {
-                    RunContext runContext = runContextFactory.of(flow, abstractTrigger);
-                    ConditionContext conditionContext = conditionService.conditionContext(runContext, flow, null);
+        flows
+            .stream()
+            .filter(flow -> flow.getTriggers() != null && !flow.getTriggers().isEmpty())
+            .flatMap(flow -> flow.getTriggers().stream().filter(trigger -> trigger instanceof PollingTriggerInterface).map(trigger -> new FlowAndTrigger(flow, trigger)))
+            .forEach(flowAndTrigger -> {
+                Optional<Trigger> trigger = triggers.stream().filter(t -> t.uid().equals(Trigger.uid(flowAndTrigger.flow(), flowAndTrigger.trigger()))).findFirst(); // must have one or none
+                if (trigger.isEmpty()) {
+                    RunContext runContext = runContextFactory.of(flowAndTrigger.flow(), flowAndTrigger.trigger());
+                    ConditionContext conditionContext = conditionService.conditionContext(runContext, flowAndTrigger.flow(), null);
                     try {
                         // new polling triggers will be evaluated immediately except schedule that will be evaluated at the next cron schedule
-                        ZonedDateTime nextExecutionDate = pollingAbstractTrigger instanceof Schedule ? pollingAbstractTrigger.nextEvaluationDate(conditionContext, Optional.empty()): now();
+                        ZonedDateTime nextExecutionDate = flowAndTrigger.trigger() instanceof Schedule  schedule ? schedule.nextEvaluationDate(conditionContext, Optional.empty()): now();
                         Trigger newTrigger = Trigger.builder()
-                            .tenantId(flow.getTenantId())
-                            .namespace(flow.getNamespace())
-                            .flowId(flow.getId())
-                            .flowRevision(flow.getRevision())
-                            .triggerId(abstractTrigger.getId())
+                            .tenantId(flowAndTrigger.flow().getTenantId())
+                            .namespace(flowAndTrigger.flow().getNamespace())
+                            .flowId(flowAndTrigger.flow().getId())
+                            .flowRevision(flowAndTrigger.flow().getRevision())
+                            .triggerId(flowAndTrigger.trigger().getId())
                             .date(now())
                             .nextExecutionDate(nextExecutionDate)
-                            .stopAfter(abstractTrigger.getStopAfter())
+                            .stopAfter(flowAndTrigger.trigger().getStopAfter())
                             .build();
                         this.triggerState.create(newTrigger);
                     } catch (Exception e) {
-                        logError(conditionContext, flow, abstractTrigger, e);
+                        logError(conditionContext, flowAndTrigger.flow(), flowAndTrigger.trigger(), e);
+                    }
+                } else if (flowAndTrigger.trigger() instanceof Schedule schedule) {
+                    // we recompute the Schedule nextExecutionDate if needed
+                    RunContext runContext = runContextFactory.of(flowAndTrigger.flow(), flowAndTrigger.trigger());
+                    Schedule.RecoverMissedSchedules recoverMissedSchedules = Optional.ofNullable(schedule.getRecoverMissedSchedules()).orElseGet(() -> schedule.defaultRecoverMissedSchedules(runContext));
+                    if (recoverMissedSchedules == Schedule.RecoverMissedSchedules.LAST) {
+                        ConditionContext conditionContext = conditionService.conditionContext(runContext, flowAndTrigger.flow(), null);
+                        ZonedDateTime previousDate = schedule.previousEvaluationDate(conditionContext);
+                        if (previousDate.isAfter(trigger.get().getDate())) {
+                            Trigger updated = trigger.get().toBuilder().nextExecutionDate(previousDate).build();
+                            this.triggerState.update(updated);
+                        }
+                    } else if (recoverMissedSchedules == Schedule.RecoverMissedSchedules.NONE ) {
+                        Trigger updated = trigger.get().toBuilder().nextExecutionDate(schedule.nextEvaluationDate()).build();
+                        this.triggerState.update(updated);
                     }
                 }
             });
-        });
+
+        this.isReady = true;
     }
 
     private List<FlowWithTriggers> computeSchedulable(List<Flow> flows, List<Trigger> triggerContextsToEvaluate, ScheduleContextInterface scheduleContext) {
@@ -230,10 +258,10 @@ public abstract class AbstractScheduler implements Scheduler {
                             .filter(triggerContextToFind -> triggerContextToFind.uid().equals(Trigger.uid(flow, abstractTrigger)))
                             .findFirst()
                             .orElse(null);
-                        // If trigger is not found in triggers to evaluate, then we ignore it
+                        // If a trigger is not found in triggers to evaluate, then we ignore it
                         if (lastTrigger == null) {
                             return null;
-                            // Backwards compatibility: we add a next execution date that we compute, this avoid re-triggering all existing trigger
+                            // Backwards compatibility: we add a next execution date that we compute, this avoids re-triggering all existing triggers
                         } else if (lastTrigger.getNextExecutionDate() == null) {
                             triggerContext = lastTrigger.toBuilder()
                                 .nextExecutionDate(((PollingTriggerInterface) abstractTrigger).nextEvaluationDate(conditionContext, Optional.of(lastTrigger)))
@@ -261,8 +289,9 @@ public abstract class AbstractScheduler implements Scheduler {
     abstract public void handleNext(List<Flow> flows, ZonedDateTime now, BiConsumer<List<Trigger>, ScheduleContextInterface> consumer);
 
     private void handle() {
-        if (!this.isReady) {
+        if (!isReady()) {
             log.warn("Scheduler is not ready, waiting");
+            return;
         }
 
         ZonedDateTime now = now();
@@ -677,7 +706,7 @@ public abstract class AbstractScheduler implements Scheduler {
     public static class FlowWithPollingTriggerNextDate extends FlowWithPollingTrigger {
         private ZonedDateTime next;
 
-        public static FlowWithPollingTriggerNextDate of(FlowWithPollingTrigger f) {
+        private static FlowWithPollingTriggerNextDate of(FlowWithPollingTrigger f) {
             return FlowWithPollingTriggerNextDate.builder()
                 .flow(f.getFlow())
                 .abstractTrigger(f.getAbstractTrigger())
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 fe738c97b..5c29ea876 100644
--- a/core/src/main/java/io/kestra/core/schedulers/DefaultScheduler.java
+++ b/core/src/main/java/io/kestra/core/schedulers/DefaultScheduler.java
@@ -40,7 +40,6 @@ public class DefaultScheduler extends AbstractScheduler {
     ) {
         super(applicationContext, flowListeners);
         this.triggerState = triggerState;
-        this.isReady = true;
 
         this.conditionService = applicationContext.getBean(ConditionService.class);
         this.flowRepository = applicationContext.getBean(FlowRepositoryInterface.class);
diff --git a/core/src/main/java/io/kestra/core/tasks/flows/ForEachItem.java b/core/src/main/java/io/kestra/core/tasks/flows/ForEachItem.java
index 493822aef..cd2985b0a 100644
--- a/core/src/main/java/io/kestra/core/tasks/flows/ForEachItem.java
+++ b/core/src/main/java/io/kestra/core/tasks/flows/ForEachItem.java
@@ -110,7 +110,65 @@ import static io.kestra.core.utils.Rethrow.throwFunction;
                     inputs:
                       order: "{{ taskrun.items }}" # special variable that contains the items of the batch
                 """
-        )
+        ),
+        @Example(
+            title = """
+                Execute a subflow for each JSON item fetched from a REST API. The subflow `mysubflow` is called from the parent flow `iterate_over_json` using the `ForEachItem` task; this creates one subflow execution for each JSON object.
+                
+                Note how we first need to convert the JSON array to JSON-L format using the `JsonWriter` task. This is because the `items` attribute of the `ForEachItem` task expects a file where each line represents a single item. Suitable file types include Amazon ION (commonly produced by Query tasks), newline-separated JSON files, or CSV files formatted with one row per line and without a header. For other formats, you can use the conversion tasks available in the `io.kestra.plugin.serdes` module.
+
+                In this example, the subflow `mysubflow` expects a JSON object as input. The `JsonReader` task first reads the JSON array from the REST API and converts it to ION. Then, the `JsonWriter` task converts that ION file to JSON-L format, suitable for the `ForEachItem` task.
+
+                ```yaml
+                id: mysubflow
+                namespace: dev
+
+                inputs:
+                  - id: json
+                    type: JSON
+
+                tasks:
+                  - id: debug
+                    type: io.kestra.core.tasks.log.Log
+                    message: "{{ inputs.json }}"
+                ```
+                """,
+            full = true,
+            code = """
+                id: iterate_over_json
+                namespace: dev
+
+                tasks:
+                  - id: download
+                    type: io.kestra.plugin.fs.http.Download
+                    uri: "https://api.restful-api.dev/objects"
+                    contentType: application/json
+                    method: GET
+                    failOnEmptyResponse: true
+                    timeout: PT15S
+
+                  - id: json_to_ion
+                    type: io.kestra.plugin.serdes.json.JsonReader
+                    from: "{{ outputs.download.uri }}"
+                    newLine: false # regular json
+
+                  - id: ion_to_jsonl
+                    type: io.kestra.plugin.serdes.json.JsonWriter
+                    from: "{{ outputs.json_to_ion.uri }}"
+                    newLine: true # JSON-L
+
+                  - id: for_each_item
+                    type: io.kestra.core.tasks.flows.ForEachItem
+                    items: "{{ outputs.ion_to_jsonl.uri }}"
+                    batch:
+                      rows: 1
+                    namespace: dev
+                    flowId: mysubflow
+                    wait: true 
+                    transmitFailed: true 
+                    inputs:
+                      json: '{{ json(read(taskrun.items)) }}'"""
+        )        
     }
 )
 public class ForEachItem extends Task implements ExecutableTask<ForEachItem.Output> {
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 10ca77e1f..3d415996f 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcScheduler.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcScheduler.java
@@ -52,7 +52,6 @@ public class JdbcScheduler extends AbstractScheduler {
         conditionService = applicationContext.getBean(ConditionService.class);
         flowRepository = applicationContext.getBean(FlowRepositoryInterface.class);
         dslContextWrapper = applicationContext.getBean(JooqDSLContextWrapper.class);
-        this.isReady = true;
     }
 
     @Override
diff --git a/ui/src/translations.json b/ui/src/translations.json
index d972d3609..952edaec9 100644
--- a/ui/src/translations.json
+++ b/ui/src/translations.json
@@ -1176,14 +1176,14 @@
     "continue backfill": "Continuer le backfill",
     "delete backfill": "Supprimer le backfill",
     "pause backfill": "Mettre en pause le backfill",
-    "backfill executions": "Backfill d'éxécutions",
-    "execute backfill": "Éxécuter le backfill",
+    "backfill executions": "Backfill d'exécutions",
+    "execute backfill": "Exécuter le backfill",
     "backfill": "Backfill",
-    "Set labels tooltip": "Ajouter des labels à l'éxécution",
+    "Set labels tooltip": "Ajouter des labels à l'exécution",
     "Set labels": "Ajouter des labels",
-    "Set labels to execution": "Ajouter ou mettre à jour des labels à l'éxécution <code>{id}</code>",
-    "Set labels done": "Labels ajoutés avec succès à l'éxécution",
-    "bulk set labels": "Etes-vous sûr de vouloir ajouter des labels à <code>{executionCount}</code>  éxécutions(s)?"
+    "Set labels to execution": "Ajouter ou mettre à jour des labels à l'exécution <code>{id}</code>",
+    "Set labels done": "Labels ajoutés avec succès à l'exécution",
+    "bulk set labels": "Etes-vous sûr de vouloir ajouter des labels à <code>{executionCount}</code>  exécutions(s)?"
   }
 }
 


Test patch:
diff --git a/core/src/test/java/io/kestra/core/models/triggers/types/ScheduleTest.java b/core/src/test/java/io/kestra/core/models/triggers/types/ScheduleTest.java
index a08e200a5..2b7543eb8 100644
--- a/core/src/test/java/io/kestra/core/models/triggers/types/ScheduleTest.java
+++ b/core/src/test/java/io/kestra/core/models/triggers/types/ScheduleTest.java
@@ -3,6 +3,7 @@ package io.kestra.core.models.triggers.types;
 import io.kestra.core.models.Label;
 import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.conditions.types.DateTimeBetweenCondition;
+import io.kestra.core.models.conditions.types.DayWeekCondition;
 import io.kestra.core.models.conditions.types.DayWeekInMonthCondition;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.Flow;
@@ -17,12 +18,10 @@ import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
 import jakarta.inject.Inject;
 import org.junit.jupiter.api.Test;
 
-import java.time.DayOfWeek;
-import java.time.Duration;
-import java.time.ZoneId;
-import java.time.ZonedDateTime;
+import java.time.*;
 import java.time.format.DateTimeFormatter;
 import java.time.temporal.ChronoUnit;
+import java.time.temporal.TemporalUnit;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -390,6 +389,8 @@ class ScheduleTest {
         assertThat(dateFromVars(vars.get("previous"), date), is(date.minusMonths(1)));
     }
 
+
+
     private ConditionContext conditionContext(AbstractTrigger trigger) {
         Flow flow = Flow.builder()
             .id(IdUtils.create())
diff --git a/core/src/test/java/io/kestra/core/plugins/PluginConfigurationTest.java b/core/src/test/java/io/kestra/core/plugins/PluginConfigurationTest.java
index a4157fae3..59dfb7130 100644
--- a/core/src/test/java/io/kestra/core/plugins/PluginConfigurationTest.java
+++ b/core/src/test/java/io/kestra/core/plugins/PluginConfigurationTest.java
@@ -9,24 +9,20 @@ import java.util.List;
 import java.util.Map;
 import java.util.stream.IntStream;
 
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.MatcherAssert.assertThat;
+
 @MicronautTest
 class PluginConfigurationTest {
 
-    private static final String PLUGIN_TEST = "io.kestra.plugin.Test";
-
     @Inject
     private List<PluginConfiguration> configurations;
 
     @Test
     void testInjectEachProperty() {
-        // Given
-        this.configurations.sort(PluginConfiguration.COMPARATOR);
-
-        // Then
-        List<PluginConfiguration> expected = IntStream.range(0, 3)
-            .mapToObj(idx -> new PluginConfiguration(idx, PLUGIN_TEST + idx, Map.of("prop" + idx, "value" + idx)))
-            .toList();
-        Assertions.assertEquals(expected, configurations);
+        assertThat(this.configurations, hasItem(new PluginConfiguration(0, "io.kestra.plugin.Test0", Map.of("prop0", "value0"))));
+        assertThat(this.configurations, hasItem(new PluginConfiguration(1, "io.kestra.plugin.Test1", Map.of("prop1", "value1"))));
+        assertThat(this.configurations, hasItem(new PluginConfiguration(2, "io.kestra.plugin.Test2", Map.of("prop2", "value2"))));
     }
 
 }
\ No newline at end of file
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 ad5b00de8..3a2bb171b 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleTest.java
@@ -111,8 +111,7 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
         triggerState.create(trigger.toBuilder().triggerId("schedule-invalid").flowId(invalid.getId()).build());
 
         // scheduler
-        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy);
-             Worker worker = new TestMethodScopedWorker(applicationContext, 8, null)) {
+        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy)) {
             // wait for execution
             Runnable assertionStop = executionQueue.receive(either -> {
                 Execution execution = either.getLeft();
@@ -135,7 +134,6 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
                 }
             });
 
-            worker.run();
             scheduler.run();
             queueCount.await(1, TimeUnit.MINUTES);
             invalidLogCount.await(1, TimeUnit.MINUTES);
@@ -173,9 +171,7 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
         triggerState.create(trigger);
 
         // scheduler
-        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy);
-             Worker worker = new TestMethodScopedWorker(applicationContext, 8, null)) {
-            worker.run();
+        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy)) {
             scheduler.run();
 
             Await.until(() -> {
@@ -187,6 +183,128 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
         }
     }
 
+    @Test
+    void recoverALLMissing() throws Exception {
+        // mock flow listeners
+        FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
+        Flow flow = createScheduleFlow("Europe/Paris", "recoverALLMissing", false);
+        doReturn(List.of(flow))
+            .when(flowListenersServiceSpy)
+            .flows();
+
+        ZonedDateTime lastDate = ZonedDateTime.now().minusHours(3L);
+        Trigger lastTrigger = Trigger
+            .builder()
+            .triggerId("recoverALLMissing")
+            .flowId(flow.getId())
+            .namespace(flow.getNamespace())
+            .date(lastDate)
+            .build();
+        triggerState.create(lastTrigger);
+
+        CountDownLatch queueCount = new CountDownLatch(1);
+
+        // scheduler
+        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy)) {
+            // wait for execution
+            Runnable assertionStop = executionQueue.receive(either -> {
+                Execution execution = either.getLeft();
+                assertThat(execution.getFlowId(), is(flow.getId()));
+                queueCount.countDown();
+            });
+
+            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));
+            Trigger newTrigger = this.triggerState.findLast(lastTrigger).orElseThrow();
+            assertThat(newTrigger.getDate().toLocalDateTime(), is(lastDate.plusHours(1L).truncatedTo(ChronoUnit.HOURS).toLocalDateTime()));
+            assertThat(newTrigger.getNextExecutionDate().toLocalDateTime(), is(lastDate.plusHours(2L).truncatedTo(ChronoUnit.HOURS).toLocalDateTime()));
+        }
+    }
+
+    @Test
+    void recoverLASTMissing() throws Exception {
+        // mock flow listeners
+        FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
+        Schedule schedule = createScheduleTrigger("Europe/Paris", "0 * * * *", "recoverLASTMissing", false)
+            .recoverMissedSchedules(Schedule.RecoverMissedSchedules.LAST)
+            .build();
+        Flow flow = createFlow(List.of(schedule));
+        doReturn(List.of(flow))
+            .when(flowListenersServiceSpy)
+            .flows();
+
+        ZonedDateTime lastDate = ZonedDateTime.now().minusHours(3L);
+        Trigger lastTrigger = Trigger
+            .builder()
+            .triggerId("recoverLASTMissing")
+            .flowId(flow.getId())
+            .namespace(flow.getNamespace())
+            .date(lastDate)
+            .build();
+        triggerState.create(lastTrigger);
+
+        CountDownLatch queueCount = new CountDownLatch(1);
+
+        // scheduler
+        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy)) {
+            // wait for execution
+            Runnable assertionStop = executionQueue.receive(either -> {
+                Execution execution = either.getLeft();
+                assertThat(execution.getFlowId(), is(flow.getId()));
+                queueCount.countDown();
+            });
+
+            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));
+            Trigger newTrigger = this.triggerState.findLast(lastTrigger).orElseThrow();
+            assertThat(newTrigger.getDate().toLocalDateTime(), is(lastDate.plusHours(3L).truncatedTo(ChronoUnit.HOURS).toLocalDateTime()));
+            assertThat(newTrigger.getNextExecutionDate().toLocalDateTime(), is(lastDate.plusHours(4L).truncatedTo(ChronoUnit.HOURS).toLocalDateTime()));
+        }
+    }
+
+    @Test
+    void recoverNONEMissing() throws Exception {
+        // mock flow listeners
+        FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
+        Schedule schedule = createScheduleTrigger("Europe/Paris", "0 * * * *", "recoverNONEMissing", false)
+            .recoverMissedSchedules(Schedule.RecoverMissedSchedules.NONE)
+            .build();
+        Flow flow = createFlow(List.of(schedule));
+        doReturn(List.of(flow))
+            .when(flowListenersServiceSpy)
+            .flows();
+
+        ZonedDateTime lastDate = ZonedDateTime.now().minusHours(3L);
+        Trigger lastTrigger = Trigger
+            .builder()
+            .triggerId("recoverNONEMissing")
+            .flowId(flow.getId())
+            .namespace(flow.getNamespace())
+            .date(lastDate)
+            .build();
+        triggerState.create(lastTrigger);
+
+        // scheduler
+        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy)) {
+            scheduler.run();
+
+            Await.until(() -> scheduler.isReady(), Duration.ofMillis(100), Duration.ofSeconds(5));
+
+            Trigger newTrigger = this.triggerState.findLast(lastTrigger).orElseThrow();
+            assertThat(newTrigger.getNextExecutionDate().toLocalDateTime(), is(lastDate.plusHours(4L).truncatedTo(ChronoUnit.HOURS).toLocalDateTime()));
+        }
+    }
+
     @Test
     void backfill() throws Exception {
         // mock flow listeners
@@ -208,9 +326,7 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
             .build();
 
         // scheduler
-        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy);
-             Worker worker = new TestMethodScopedWorker(applicationContext, 8, null)) {
-            worker.run();
+        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy)) {
             scheduler.run();
 
             Await.until(() -> {
@@ -275,13 +391,11 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
         triggerState.create(trigger);
 
         // scheduler
-        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy);
-             Worker worker = new TestMethodScopedWorker(applicationContext, 8, null)) {
-            worker.run();
+        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy)) {
             scheduler.run();
 
-            // Wait 5s to see if things happen
-            Thread.sleep(5000);
+            // Wait 3s to see if things happen
+            Thread.sleep(3000);
 
             Trigger lastTrigger = this.triggerState.findLast(trigger).get();
 
@@ -315,8 +429,7 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
         CountDownLatch queueCount = new CountDownLatch(2);
 
         // scheduler
-        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy);
-             Worker worker = new TestMethodScopedWorker(applicationContext, 8, null)) {
+        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy)) {
             // wait for execution
             Runnable assertionStop = executionQueue.receive(either -> {
                 Execution execution = either.getLeft();
@@ -330,7 +443,6 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
                 }
             });
 
-            worker.run();
             scheduler.run();
 
             queueCount.await(1, TimeUnit.MINUTES);
@@ -375,11 +487,10 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
             .build();
         triggerState.create(lastTrigger);
 
-        CountDownLatch queueCount = new CountDownLatch(2);
+        CountDownLatch queueCount = new CountDownLatch(1);
 
         // scheduler
-        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy);
-             Worker worker = new TestMethodScopedWorker(applicationContext, 8, null)) {
+        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy)) {
             // wait for execution
             Runnable assertionStop = executionQueue.receive(either -> {
                 Execution execution = either.getLeft();
@@ -389,7 +500,6 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
                 queueCount.countDown();
             });
 
-            worker.run();
             scheduler.run();
 
             queueCount.await(1, TimeUnit.MINUTES);
diff --git a/core/src/test/resources/application-test.yml b/core/src/test/resources/application-test.yml
index 6547cef2f..7246e1347 100644
--- a/core/src/test/resources/application-test.yml
+++ b/core/src/test/resources/application-test.yml
@@ -54,4 +54,7 @@ kestra:
           prop1: value1
       - type: io.kestra.plugin.Test2
         values:
-          prop2: value2
\ No newline at end of file
+          prop2: value2
+      - type: io.kestra.core.models.triggers.types.Schedule
+        values:
+          recover-missed-schedules: ALL
\ No newline at end of file