definitely worth showing a warning in the UI when creating a flow with such invalid condition and disable those WARN logs.

I was able to reproduce that this log message appears every couple of milliseconds:

```Java
2024-01-07 20:07:46,922 WARN  pool-7-thread-1 f.h.naive-schedule-condition [namespace: company.team] [flow: hello-world-var-condition] [condition: VariableCondition(super=io.kestra.core.models.conditions.types.VariableCondition@596bcd, expression={{ trigger.date | date() < now() }})] Evaluate Condition Failed with error 'io.pebbletemplates.pebble.error.PebbleException: Could not perform greater modulus ({{ trigger.date | date() < now() }}:1)'
io.kestra.core.exceptions.IllegalVariableEvaluationException: io.pebbletemplates.pebble.error.PebbleException: Could not perform greater modulus ({{ trigger.date | date() < now() }}:1)
	at io.kestra.core.runners.VariableRenderer.properPebbleException(VariableRenderer.java:166)
	at io.kestra.core.runners.VariableRenderer.recursiveRender(VariableRenderer.java:127)
	at io.kestra.core.runners.VariableRenderer.render(VariableRenderer.java:170)
	at io.kestra.core.runners.RunContext.render(RunContext.java:442)
	at io.kestra.core.models.conditions.types.VariableCondition.test(VariableCondition.java:50)
	at io.kestra.core.services.ConditionService.lambda$isValid$0(ConditionService.java:69)
	at java.base/java.util.stream.MatchOps$1MatchSink.accept(Unknown Source)
	at java.base/java.util.ArrayList$ArrayListSpliterator.tryAdvance(Unknown Source)
	at java.base/java.util.stream.ReferencePipeline.forEachWithCancel(Unknown Source)
	at java.base/java.util.stream.AbstractPipeline.copyIntoWithCancel(Unknown Source)
	at java.base/java.util.stream.AbstractPipeline.copyInto(Unknown Source)
	at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(Unknown Source)
	at java.base/java.util.stream.MatchOps$MatchOp.evaluateSequential(Unknown Source)
	at java.base/java.util.stream.MatchOps$MatchOp.evaluateSequential(Unknown Source)
	at java.base/java.util.stream.AbstractPipeline.evaluate(Unknown Source)
	at java.base/java.util.stream.ReferencePipeline.allMatch(Unknown Source)
	at io.kestra.core.services.ConditionService.isValid(ConditionService.java:67)
	at io.kestra.core.models.triggers.types.Schedule.validateScheduleCondition(Schedule.java:433)
	at io.kestra.core.models.triggers.types.Schedule.truePreviousNextDateWithCondition(Schedule.java:393)
	at io.kestra.core.models.triggers.types.Schedule.nextEvaluationDate(Schedule.java:186)
	at io.kestra.core.schedulers.AbstractScheduler.lambda$handle$15(AbstractScheduler.java:259)
	at java.base/java.util.stream.ReferencePipeline$3$1.accept(Unknown Source)
	at java.base/java.util.stream.ReferencePipeline$2$1.accept(Unknown Source)
	at java.base/java.util.stream.ReferencePipeline$2$1.accept(Unknown Source)
	at java.base/java.util.stream.ReferencePipeline$2$1.accept(Unknown Source)
	at java.base/java.util.stream.ReferencePipeline$3$1.accept(Unknown Source)
	at java.base/java.util.stream.ReferencePipeline$2$1.accept(Unknown Source)
	at java.base/java.util.AbstractList$RandomAccessSpliterator.forEachRemaining(Unknown Source)
	at java.base/java.util.stream.AbstractPipeline.copyInto(Unknown Source)
	at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(Unknown Source)
	at java.base/java.util.stream.AbstractPipeline.evaluate(Unknown Source)
	at java.base/java.util.stream.AbstractPipeline.evaluateToArrayNode(Unknown Source)
	at java.base/java.util.stream.ReferencePipeline.toArray(Unknown Source)
	at java.base/java.util.stream.ReferencePipeline.toArray(Unknown Source)
	at java.base/java.util.stream.ReferencePipeline.toList(Unknown Source)
	at io.kestra.core.schedulers.AbstractScheduler.handle(AbstractScheduler.java:268)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
	at java.base/java.util.concurrent.FutureTask.runAndReset(Unknown Source)
	at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(Unknown Source)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
	at java.base/java.lang.Thread.run(Unknown Source)
Caused by: io.pebbletemplates.pebble.error.PebbleException: Could not perform greater modulus ({{ trigger.date | date() < now() }}:1)
	at io.pebbletemplates.pebble.node.expression.LessThanExpression.evaluate(LessThanExpression.java:25)
	at io.pebbletemplates.pebble.node.expression.LessThanExpression.evaluate(LessThanExpression.java:16)
	at io.pebbletemplates.pebble.node.PrintNode.render(PrintNode.java:37)
	at io.pebbletemplates.pebble.node.BodyNode.render(BodyNode.java:44)
	at io.pebbletemplates.pebble.node.RootNode.render(RootNode.java:31)
	at io.pebbletemplates.pebble.template.PebbleTemplateImpl.evaluate(PebbleTemplateImpl.java:157)
	at io.pebbletemplates.pebble.template.PebbleTemplateImpl.evaluate(PebbleTemplateImpl.java:96)
	at io.kestra.core.runners.VariableRenderer.recursiveRender(VariableRenderer.java:122)
	... 40 common frames omitted
Caused by: java.lang.RuntimeException: invalid operands for mathematical comparison [LESS_THAN]
	at io.pebbletemplates.pebble.utils.OperatorUtils.wideningConversionBinaryComparison(OperatorUtils.java:189)
	at io.pebbletemplates.pebble.utils.OperatorUtils.lt(OperatorUtils.java:94)
	at io.pebbletemplates.pebble.node.expression.LessThanExpression.evaluate(LessThanExpression.java:21)
	... 47 common frames omitted
``` - change the WARN level to TRACE or DEBUG
- add a short and informative error message that Pebble in the condition is invalid
Using invalid Pebble syntax within a schedule trigger leads to excessive logging
### Explain the bug

Using an invalid `VariableCondition` within a schedule trigger leads to a continuous stream of Pebble's `java.lang.RuntimeException`. This leads to excessive logging with related implications.

---

Saving the following flow invokes such exception featuring the `invalid operands for mathematical comparison [LESS_THAN]` message.

```yaml
id: hello-world-var-condition
namespace: company.team

triggers:
  - id: naive-schedule-condition
    type: io.kestra.core.models.triggers.types.Schedule
    description: Schedule featuring an invalid `VariableCondition`
    cron: "*/5 * * * * "
    scheduleConditions:
     - type: io.kestra.core.models.conditions.types.VariableCondition
       expression: "{{ trigger.date | date() < now() }}" # FIXME: invalid Pebble syntax leads to an extreme log spam

tasks:
  - id: hello
    type: io.kestra.core.tasks.log.Log
    message: Kestra team wishes you a great day! 👋
```

### Environment Information

- Kestra Version: v0.14.0-SNAPSHOT as of Jan 5th 2024
- Operating System and Java Version (if not using Kestra Docker image): Corretto 17

core/src/main/java/io/kestra/core/models/triggers/types/Schedule.java core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java core/src/main/java/io/kestra/core/services/ConditionService.java
core/src/main/java/io/kestra/core/models/triggers/types/Schedule.java core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java core/src/main/java/io/kestra/core/services/ConditionService.java
core/src/test/java/io/kestra/core/schedulers/SchedulerPollingTriggerTest.java core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleTest.java
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 78d29fa4c4..727a71062a 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
@@ -5,6 +5,7 @@
 import com.cronutils.model.time.ExecutionTime;
 import com.cronutils.parser.CronParser;
 import com.google.common.collect.ImmutableMap;
+import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.models.Label;
 import io.kestra.core.models.annotations.Example;
 import io.kestra.core.models.annotations.Plugin;
@@ -87,7 +88,6 @@
                 "  - id: schedule",
                 "    cron: \"0 11 * * 1\"",
                 "    scheduleConditions:",
-                "      - id: monday",
                 "        type: io.kestra.core.models.conditions.types.DayWeekInMonthCondition",
                 "        date: \"{{ trigger.date }}\"",
                 "        dayOfWeek: \"MONDAY\"",
@@ -191,16 +191,20 @@ public ZonedDateTime nextEvaluationDate(ConditionContext conditionContext, Optio
             }
             // previous present & scheduleConditions
             if (this.scheduleConditions != null) {
-                Optional<ZonedDateTime> next = this.truePreviousNextDateWithCondition(
-                    executionTime,
-                    conditionContext,
-                    lastDate,
-                    true
-                );
-                if (next.isPresent()) {
-
-                    return next.get().truncatedTo(ChronoUnit.SECONDS);
+                try {
+                    Optional<ZonedDateTime>next = this.truePreviousNextDateWithCondition(
+                        executionTime,
+                        conditionContext,
+                        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);
@@ -268,9 +272,26 @@ public Optional<Execution> evaluate(ConditionContext conditionContext, TriggerCo
         // FIXME make scheduleConditions generic
         // control scheduleConditions
         if (scheduleConditions != null) {
-            boolean conditionResults = this.validateScheduleCondition(conditionContext);
-            if (!conditionResults) {
-                return Optional.empty();
+            try {
+                boolean conditionResults = this.validateScheduleCondition(conditionContext);
+                if (!conditionResults) {
+                    return Optional.empty();
+                }
+            } catch(InternalException ie) {
+                // validate schedule condition can fail to render variables
+                // in this case, we return a failed execution so the trigger is not evaluated each second
+                runContext.logger().error("Unable to evaluate the Schedule trigger '{}'", this.getId(), ie);
+                List<Label> labels = getLabels(conditionContext, backfill);
+                Execution execution = Execution.builder()
+                    .id(runContext.getTriggerExecutionId())
+                    .tenantId(triggerContext.getTenantId())
+                    .namespace(triggerContext.getNamespace())
+                    .flowId(triggerContext.getFlowId())
+                    .flowRevision(triggerContext.getFlowRevision())
+                    .labels(labels)
+                    .state(new State().withState(State.Type.FAILED))
+                    .build();
+                return Optional.of(execution);
             }
 
             // recalculate true output for previous and next based on conditions
@@ -301,10 +322,7 @@ public Optional<Execution> evaluate(ConditionContext conditionContext, TriggerCo
         } else {
             variables = scheduleDates.toMap();
         }
-        List<Label> labels = conditionContext.getFlow().getLabels() != null ? conditionContext.getFlow().getLabels() : new ArrayList<>();
-        if (backfill != null && backfill.getLabels() != null) {
-            labels.addAll(backfill.getLabels());
-        }
+        List<Label> labels = getLabels(conditionContext, backfill);
 
         ExecutionTrigger executionTrigger = ExecutionTrigger.of(this, variables);
 
@@ -332,6 +350,14 @@ public Optional<Execution> evaluate(ConditionContext conditionContext, TriggerCo
         return Optional.of(execution);
     }
 
+    private static 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());
+        }
+        return labels;
+    }
+
     private Optional<Output> scheduleDates(ExecutionTime executionTime, ZonedDateTime date) {
         Optional<ZonedDateTime> next = executionTime.nextExecution(date.minus(Duration.ofSeconds(1)));
 
@@ -384,7 +410,7 @@ private Optional<ZonedDateTime> computeNextEvaluationDate(ExecutionTime executio
         return executionTime.nextExecution(date).map(zonedDateTime -> zonedDateTime.truncatedTo(ChronoUnit.SECONDS));
     }
 
-    private Output trueOutputWithCondition(ExecutionTime executionTime, ConditionContext conditionContext, Output output) {
+    private Output trueOutputWithCondition(ExecutionTime executionTime, ConditionContext conditionContext, Output output) throws InternalException {
         Output.OutputBuilder<?, ?> outputBuilder = Output.builder()
             .date(output.getDate());
 
@@ -397,7 +423,7 @@ private Output trueOutputWithCondition(ExecutionTime executionTime, ConditionCon
         return outputBuilder.build();
     }
 
-    private Optional<ZonedDateTime> truePreviousNextDateWithCondition(ExecutionTime executionTime, ConditionContext conditionContext, ZonedDateTime toTestDate, boolean next) {
+    private Optional<ZonedDateTime> truePreviousNextDateWithCondition(ExecutionTime executionTime, ConditionContext conditionContext, ZonedDateTime toTestDate, boolean next) throws InternalException {
         while (
             (next && toTestDate.getYear() < ZonedDateTime.now().getYear() + 10) ||
                 (!next && toTestDate.getYear() > ZonedDateTime.now().getYear() - 10)
@@ -455,11 +481,10 @@ private Output handleMaxDelay(Output output) {
         return output;
     }
 
-    private boolean validateScheduleCondition(ConditionContext conditionContext) {
+    private boolean validateScheduleCondition(ConditionContext conditionContext) throws InternalException {
         if (scheduleConditions != null) {
             ConditionService conditionService = conditionContext.getRunContext().getApplicationContext().getBean(ConditionService.class);
             return conditionService.isValid(
-                conditionContext.getFlow(),
                 scheduleConditions,
                 conditionContext
             );
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 584d811adc..238e7d77ec 100644
--- a/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
+++ b/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
@@ -7,6 +7,7 @@
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.FlowWithException;
+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;
@@ -357,13 +358,17 @@ private void handle() {
                                 e
                             );
                         }
-                    } else if (f.getPollingTrigger() instanceof Schedule) {
+                    } else if (f.getPollingTrigger() instanceof Schedule schedule) {
                         // This is the Schedule, all other triggers should have an interval.
                         // So we evaluate it now as there is no need to send it to the worker.
                         // Schedule didn't use the triggerState to allow backfill.
                         try {
-                            SchedulerExecutionWithTrigger schedulerExecutionWithTrigger = evaluateScheduleTrigger(f);
-                            this.handleEvaluateSchedulingTriggerResult(schedulerExecutionWithTrigger, scheduleContext);
+                            Optional<SchedulerExecutionWithTrigger> maybe = evaluateScheduleTrigger(f);
+                            maybe.ifPresent(
+                                schedulerExecutionWithTrigger -> this.handleEvaluateSchedulingTriggerResult(schedule, schedulerExecutionWithTrigger, f.getConditionContext(), scheduleContext)
+                            );
+                            // TODO maybe compute the next date and save the trigger as Schedule are evaluated each second if evaluate didn't leads to an execution
+
                         } catch (Exception e) {
                             logger.error(
                                 "[namespace: {}] [flow: {}] [trigger: {}] Evaluate schedule trigger failed",
@@ -402,7 +407,7 @@ private void handleEvaluatePollingTriggerResult(SchedulerExecutionWithTrigger re
 
                     // Check if the localTriggerState contains it
                     // however, its mean it has been deleted during the execution time
-                    this.triggerState.update(trigger);
+                    this.triggerState.update(trigger); //FIXME should be moved to saveLastTriggerAndEmitExecution as it creates duplicate emission in the Trigger topic in Kafka
                     this.saveLastTriggerAndEmitExecution(executionWithTrigger, trigger);
                 }
             );
@@ -411,21 +416,22 @@ private void handleEvaluatePollingTriggerResult(SchedulerExecutionWithTrigger re
     // Schedule triggers are being executed directly from
     // the handle method within the context where triggers are locked,
     // so we can save it now by pass the context
-    private void handleEvaluateSchedulingTriggerResult(SchedulerExecutionWithTrigger result, ScheduleContextInterface scheduleContext) {
-        Stream.of(result)
-            .filter(Objects::nonNull)
-            .peek(this::log)
-            .forEach(executionWithTrigger -> {
-                    Trigger trigger = Trigger.of(
-                        executionWithTrigger.getTriggerContext(),
-                        executionWithTrigger.getExecution(),
-                        ZonedDateTime.parse((String) executionWithTrigger.getExecution().getTrigger().getVariables().get("next"))
-                    );
-                    trigger = trigger.checkBackfill();
-                    this.triggerState.save(trigger, scheduleContext);
-                    this.saveLastTriggerAndEmitExecution(executionWithTrigger, trigger);
-                }
-            );
+    private void handleEvaluateSchedulingTriggerResult(Schedule schedule, SchedulerExecutionWithTrigger result, ConditionContext conditionContext, ScheduleContextInterface scheduleContext) {
+        log(result);
+        Trigger trigger = Trigger.of(
+            result.getTriggerContext(),
+            result.getExecution(),
+            schedule.nextEvaluationDate(conditionContext, Optional.of(result.getTriggerContext()))
+        );
+        trigger = trigger.checkBackfill();
+
+        // if the execution is already failed due to failed execution, we reset the trigger now
+        if (result.getExecution().getState().getCurrent() == State.Type.FAILED) {
+            trigger = trigger.resetExecution(State.Type.FAILED);
+        }
+
+        this.triggerState.save(trigger, scheduleContext); //FIXME should be moved to saveLastTriggerAndEmitExecution as it creates duplicate emission in the Trigger topic in Kafka
+        this.saveLastTriggerAndEmitExecution(result, trigger);
     }
 
     protected void saveLastTriggerAndEmitExecution(SchedulerExecutionWithTrigger executionWithTrigger, Trigger trigger) {
@@ -525,7 +531,7 @@ private static ZonedDateTime now() {
         return ZonedDateTime.now().truncatedTo(ChronoUnit.SECONDS);
     }
 
-    private SchedulerExecutionWithTrigger evaluateScheduleTrigger(FlowWithPollingTrigger flowWithTrigger) {
+    private Optional<SchedulerExecutionWithTrigger> evaluateScheduleTrigger(FlowWithPollingTrigger flowWithTrigger) {
         try {
             FlowWithPollingTrigger flowWithPollingTrigger = flowWithTrigger.from(taskDefaultService.injectDefaults(
                 flowWithTrigger.getFlow(),
@@ -561,10 +567,10 @@ private SchedulerExecutionWithTrigger evaluateScheduleTrigger(FlowWithPollingTri
             return evaluate.map(execution -> new SchedulerExecutionWithTrigger(
                 execution,
                 flowWithTrigger.getTriggerContext()
-            )).orElse(null);
+            ));
         } catch (Exception e) {
             logError(flowWithTrigger, e);
-            return null;
+            return Optional.empty();
         }
     }
 
diff --git a/core/src/main/java/io/kestra/core/services/ConditionService.java b/core/src/main/java/io/kestra/core/services/ConditionService.java
index 07e7ceb89a..74f885e882 100644
--- a/core/src/main/java/io/kestra/core/services/ConditionService.java
+++ b/core/src/main/java/io/kestra/core/services/ConditionService.java
@@ -1,6 +1,7 @@
 package io.kestra.core.services;
 
 import com.cronutils.utils.VisibleForTesting;
+import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.models.conditions.Condition;
 import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.conditions.ScheduleCondition;
@@ -20,6 +21,8 @@
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 
+import static io.kestra.core.utils.Rethrow.throwPredicate;
+
 /**
  * Provides business logic to manipulate {@link Condition}
  */
@@ -61,18 +64,10 @@ public boolean isValid(Flow flow, AbstractTrigger trigger, ConditionContext cond
         return this.valid(flow, conditions, conditionContext);
     }
 
-    public boolean isValid(Flow flow, List<ScheduleCondition> conditions, ConditionContext conditionContext) {
+    public boolean isValid(List<ScheduleCondition> conditions, ConditionContext conditionContext) throws InternalException {
         return conditions
             .stream()
-            .allMatch(condition -> {
-                try {
-                    return condition.test(conditionContext);
-                } catch (Exception e) {
-                    logException(flow, condition, conditionContext, e);
-
-                    return false;
-                }
-            });
+            .allMatch(throwPredicate(condition -> condition.test(conditionContext)));
     }
 
     public boolean isValid(AbstractTrigger trigger, Flow flow, Execution execution, MultipleConditionStorageInterface multipleConditionStorage) {

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 17e518d409..cfda5638ec 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerPollingTriggerTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerPollingTriggerTest.java
@@ -121,7 +121,7 @@ void pollingTriggerStopAfter() throws Exception {
             // 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));
+            Await.until(() -> this.triggerState.findLast(trigger).map(t -> t.getDisabled()).orElse(false).booleanValue(), Duration.ofMillis(100), Duration.ofSeconds(10));
         }
     }
 
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 7555f02590..76b540a109 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleTest.java
@@ -1,5 +1,6 @@
 package io.kestra.core.schedulers;
 
+import io.kestra.core.models.conditions.types.VariableCondition;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.LogEntry;
 import io.kestra.core.models.flows.Flow;
@@ -293,7 +294,9 @@ void disabled() throws Exception {
     void stopAfterSchedule() throws Exception {
         // mock flow listeners
         FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
-        Schedule schedule = createScheduleTrigger("Europe/Paris", "* * * * *", "stopAfter", false).build();
+        Schedule schedule = createScheduleTrigger("Europe/Paris", "* * * * *", "stopAfter", false)
+            .stopAfter(List.of(State.Type.SUCCESS))
+            .build();
         Flow flow = createFlow(Collections.singletonList(schedule));
         doReturn(List.of(flow))
             .when(flowListenersServiceSpy)
@@ -339,7 +342,63 @@ void stopAfterSchedule() throws Exception {
             // 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));
+            Await.until(() -> this.triggerState.findLast(trigger).map(t -> t.getDisabled()).orElse(false).booleanValue(), Duration.ofMillis(100), Duration.ofSeconds(10));
+        }
+    }
+
+    @Test
+    void failedEvaluationTest() {
+        // mock flow listeners
+        FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
+        Schedule schedule = createScheduleTrigger("Europe/Paris", "* * * * *", "failedEvaluation", false)
+            .scheduleConditions(
+                List.of(
+                    VariableCondition.builder()
+                        .type(VariableCondition.class.getName())
+                        .expression("{{ trigger.date | date() < now() }}")
+                        .build()
+                )
+            )
+            .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("failedEvaluation")
+            .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.getFlowId(), is(flow.getId()));
+                assertThat(execution.getState().getCurrent(), is(State.Type.FAILED));
+
+                queueCount.countDown();
+            });
+
+            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));
+        } catch (Exception e) {
+            throw new RuntimeException(e);
         }
     }
 }
