Invalid condition on a trigger makes the trigger being evaluated each second
### Describe the issue

Same kind of issue than https://github.com/kestra-io/kestra/issues/2811.

Example flow:

```yaml
id: hello-polling-trigger-condition
namespace: company.team

triggers:
  - id: poling-trigger
    type: io.kestra.plugin.fs.http.Trigger
    uri: https://api.chucknorris.io/jokes/random
    responseCondition: "{{ response.statusCode == 200 }}" 
    conditions:
     - type: io.kestra.core.models.conditions.types.VariableCondition
       expression: "{{ trigger.date | date() < now() }}"

tasks:
  - id: hello
    type: io.kestra.core.tasks.log.Log
    message: Polling trigger
```

### Environment

- Kestra Version:
- Operating System (OS/Docker/Kubernetes):
- Java Version (if you don't run kestra in Docker):

core/src/main/java/io/kestra/core/models/triggers/Trigger.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/Trigger.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
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 5ca748c719..216a317bc9 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
@@ -206,6 +206,10 @@ public static Trigger update(Trigger currentTrigger, Trigger newTrigger, ZonedDa
     }
 
     public Trigger resetExecution(State.Type executionEndState) {
+        return resetExecution(executionEndState, this.getNextExecutionDate());
+    }
+
+    public Trigger resetExecution(State.Type executionEndState, ZonedDateTime nextExecutionDate) {
         // switch disabled automatically if the executionEndState is one of the stopAfter states
         Boolean disabled = this.getStopAfter() != null ? this.getStopAfter().contains(executionEndState) : this.getDisabled();
 
@@ -216,7 +220,7 @@ public Trigger resetExecution(State.Type executionEndState) {
             .flowRevision(this.getFlowRevision())
             .triggerId(this.getTriggerId())
             .date(this.getDate())
-            .nextExecutionDate(this.getNextExecutionDate())
+            .nextExecutionDate(nextExecutionDate)
             .stopAfter(this.getStopAfter())
             .backfill(this.getBackfill())
             .disabled(disabled)
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 dd694694d7..0baff51383 100644
--- a/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
+++ b/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
@@ -3,6 +3,7 @@
 import com.google.common.base.Throwables;
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.metrics.MetricRegistry;
+import io.kestra.core.models.conditions.Condition;
 import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.Flow;
@@ -18,6 +19,7 @@
 import io.kestra.core.runners.*;
 import io.kestra.core.services.*;
 import io.kestra.core.utils.Await;
+import io.kestra.core.utils.IdUtils;
 import io.kestra.core.utils.ListUtils;
 import io.micronaut.context.ApplicationContext;
 import io.micronaut.inject.qualifiers.Qualifiers;
@@ -294,7 +296,6 @@ private void handle() {
             // Get all triggers that are ready for evaluation
             List<FlowWithPollingTriggerNextDate> readyForEvaluate = schedulable
                 .stream()
-                .filter(f -> conditionService.isValid(f.getFlow(), f.getAbstractTrigger(), f.getConditionContext()))
                 .map(flowWithTriggers -> FlowWithPollingTrigger.builder()
                     .flow(flowWithTriggers.getFlow())
                     .abstractTrigger(flowWithTriggers.getAbstractTrigger())
@@ -335,60 +336,72 @@ private void handle() {
                 .forEach(f -> {
                     schedulableNextDate.put(f.getTriggerContext().uid(), f);
                     Logger logger = f.getConditionContext().getRunContext().logger();
-
-                    if (f.getPollingTrigger().getInterval() != null) {
-                        // If have an interval, the trigger is executed by the Worker.
-                        // Normally, only the Schedule trigger has no interval.
-                        Trigger triggerRunning = Trigger.of(f.getTriggerContext(), now);
-
-                        try {
-                            this.triggerState.save(triggerRunning, scheduleContext);
-                            this.sendPollingTriggerToWorker(f);
-                        } catch (InternalException e) {
-                            logService.logTrigger(
-                                f.getTriggerContext(),
-                                logger,
-                                Level.ERROR,
-                                "Unable to send polling trigger to worker",
-                                e
-                            );
-                        }
-                    } 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 {
-                            Optional<SchedulerExecutionWithTrigger> schedulerExecutionWithTrigger = evaluateScheduleTrigger(f);
-                            if (schedulerExecutionWithTrigger.isPresent()) {
-                                this.handleEvaluateSchedulingTriggerResult(schedule, schedulerExecutionWithTrigger.get(), f.getConditionContext(), scheduleContext);
+                    try {
+                        // conditionService.areValid can fail, so we cannot execute it early as we need to try/catch and send a failed executions
+                        List<Condition> conditions = f.getAbstractTrigger().getConditions() != null ? f.getAbstractTrigger().getConditions() : Collections.emptyList();
+                        boolean shouldEvaluate = conditionService.areValid(conditions, f.getConditionContext());
+                        if (shouldEvaluate) {
+
+                            if (f.getPollingTrigger().getInterval() != null) {
+                                // If it has an interval, the Worker will execute the trigger.
+                                // Normally, only the Schedule trigger has no interval.
+                                Trigger triggerRunning = Trigger.of(f.getTriggerContext(), now);
+
+                                try {
+                                    this.triggerState.save(triggerRunning, scheduleContext);
+                                    this.sendPollingTriggerToWorker(f);
+                                } catch (InternalException e) {
+                                    logService.logTrigger(
+                                        f.getTriggerContext(),
+                                        logger,
+                                        Level.ERROR,
+                                        "Unable to send polling trigger to worker",
+                                        e
+                                    );
+                                }
+                            } 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.
+                                Optional<SchedulerExecutionWithTrigger> schedulerExecutionWithTrigger = evaluateScheduleTrigger(f);
+                                if (schedulerExecutionWithTrigger.isPresent()) {
+                                    this.handleEvaluateSchedulingTriggerResult(schedule, schedulerExecutionWithTrigger.get(), f.getConditionContext(), scheduleContext);
+                                } else {
+                                    // compute next date and save the trigger to avoid evaluating it each second
+                                    Trigger trigger = Trigger.fromEvaluateFailed(
+                                        f.getTriggerContext(),
+                                        schedule.nextEvaluationDate(f.getConditionContext(), Optional.of(f.getTriggerContext()))
+                                    );
+                                    trigger = trigger.checkBackfill();
+                                    this.triggerState.save(trigger, scheduleContext);
+                                }
                             } else {
-                                // compute next date and save the trigger to avoid evaluating it each second
-                                Trigger trigger = Trigger.fromEvaluateFailed(
+                                logService.logTrigger(
                                     f.getTriggerContext(),
-                                    schedule.nextEvaluationDate(f.getConditionContext(), Optional.of(f.getTriggerContext()))
+                                    logger,
+                                    Level.ERROR,
+                                    "Polling trigger must have an interval (except the Schedule)"
                                 );
-                                trigger = trigger.checkBackfill();
-                                this.triggerState.save(trigger, scheduleContext);
                             }
-                        } catch (Exception e) {
-                            logService.logTrigger(
-                                f.getTriggerContext(),
-                                logger,
-                                Level.ERROR,
-                                "Evaluate schedule trigger failed",
-                                e
-                            );
                         }
-                    } else {
-                        logService.logTrigger(
-                            f.getTriggerContext(),
-                            logger,
-                            Level.ERROR,
-                            "Polling trigger must have an interval (except the Schedule)"
-                        );
+                    } catch(InternalException ie) {
+                        // validate schedule condition can fail to render variables
+                        // in this case, we send a failed execution so the trigger is not evaluated each second.
+                        logger.error("Unable to evaluate the trigger '{}'", f.getAbstractTrigger().getId(), ie);
+                        Execution execution = Execution.builder()
+                            .id(IdUtils.create())
+                            .tenantId(f.getTriggerContext().getTenantId())
+                            .namespace(f.getTriggerContext().getNamespace())
+                            .flowId(f.getTriggerContext().getFlowId())
+                            .flowRevision(f.getTriggerContext().getFlowRevision())
+                            .labels(f.getFlow().getLabels())
+                            .state(new State().withState(State.Type.FAILED))
+                            .build();
+                        ZonedDateTime nextExecutionDate = f.getPollingTrigger().nextEvaluationDate();
+                        var trigger = f.getTriggerContext().resetExecution(State.Type.FAILED, nextExecutionDate);
+                        this.saveLastTriggerAndEmitExecution(execution, trigger, triggerToSave -> this.triggerState.save(triggerToSave, scheduleContext));
                     }
                 });
-
         });
     }
 
@@ -405,7 +418,7 @@ private void handleEvaluatePollingTriggerResult(SchedulerExecutionWithTrigger re
 
                     // Polling triggers result is evaluated in another thread with the workerTriggerResultQueue.
                     // We can then update the trigger directly.
-                    this.saveLastTriggerAndEmitExecution(executionWithTrigger, trigger, triggerToSave -> this.triggerState.update(triggerToSave));
+                    this.saveLastTriggerAndEmitExecution(executionWithTrigger.getExecution(), trigger, triggerToSave -> this.triggerState.update(triggerToSave));
                 }
             );
     }
@@ -426,15 +439,15 @@ private void handleEvaluateSchedulingTriggerResult(Schedule schedule, SchedulerE
 
         // Schedule triggers are being executed directly from the handle method within the context where triggers are locked.
         // So we must save them by passing the scheduleContext.
-        this.saveLastTriggerAndEmitExecution(result, trigger, triggerToSave -> this.triggerState.save(triggerToSave, scheduleContext));
+        this.saveLastTriggerAndEmitExecution(result.getExecution(), trigger, triggerToSave -> this.triggerState.save(triggerToSave, scheduleContext));
     }
 
-    protected void saveLastTriggerAndEmitExecution(SchedulerExecutionWithTrigger executionWithTrigger, Trigger trigger, Consumer<Trigger> saveAction) {
+    protected void saveLastTriggerAndEmitExecution(Execution execution, Trigger trigger, Consumer<Trigger> saveAction) {
         saveAction.accept(trigger);
 
         // we need to be sure that the tenantId is propagated from the trigger to the execution
-        var execution = executionWithTrigger.getExecution().withTenantId(executionWithTrigger.getTriggerContext().getTenantId());
-        this.executionQueue.emit(execution);
+        var newExecution = execution.withTenantId(trigger.getTenantId());
+        this.executionQueue.emit(newExecution);
     }
 
     private boolean isExecutionNotRunning(FlowWithPollingTrigger f) {
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 74f885e882..00cb8024cc 100644
--- a/core/src/main/java/io/kestra/core/services/ConditionService.java
+++ b/core/src/main/java/io/kestra/core/services/ConditionService.java
@@ -64,12 +64,26 @@ public boolean isValid(Flow flow, AbstractTrigger trigger, ConditionContext cond
         return this.valid(flow, conditions, conditionContext);
     }
 
+    /**
+     * Check that all conditions are valid.
+     * Warning, this method throws if a condition cannot be evaluated.
+     */
     public boolean isValid(List<ScheduleCondition> conditions, ConditionContext conditionContext) throws InternalException {
         return conditions
             .stream()
             .allMatch(throwPredicate(condition -> condition.test(conditionContext)));
     }
 
+    /**
+     * Check that all conditions are valid.
+     * Warning, this method throws if a condition cannot be evaluated.
+     */
+    public boolean areValid(List<Condition> conditions, ConditionContext conditionContext) throws InternalException {
+        return conditions
+            .stream()
+            .allMatch(throwPredicate(condition -> condition.test(conditionContext)));
+    }
+
     public boolean isValid(AbstractTrigger trigger, Flow flow, Execution execution, MultipleConditionStorageInterface multipleConditionStorage) {
         assert execution != null;
 

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 cfda5638ec..5fc9851496 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerPollingTriggerTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerPollingTriggerTest.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.flows.Flow;
 import io.kestra.core.models.flows.State;
@@ -27,7 +28,6 @@
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.spy;
 
-@MicronautTest(transactional = false, rebuildContext = true) // without 'rebuildContext = true' the second test fail
 public class SchedulerPollingTriggerTest extends AbstractSchedulerTest {
     @Inject
     private ApplicationContext applicationContext;
@@ -43,7 +43,7 @@ public class SchedulerPollingTriggerTest extends AbstractSchedulerTest {
     void pollingTrigger() throws Exception {
         // mock flow listener
         FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
-        PollingTrigger pollingTrigger = createPollingTrigger(null);
+        PollingTrigger pollingTrigger = createPollingTrigger(null).build();
         Flow flow = createPollingTriggerFlow(pollingTrigger);
         doReturn(List.of(flow))
             .when(flowListenersServiceSpy)
@@ -80,7 +80,7 @@ void pollingTrigger() throws Exception {
     void pollingTriggerStopAfter() throws Exception {
         // mock flow listener
         FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
-        PollingTrigger pollingTrigger = createPollingTrigger(List.of(State.Type.FAILED));
+        PollingTrigger pollingTrigger = createPollingTrigger(List.of(State.Type.FAILED)).build();
         Flow flow = createPollingTriggerFlow(pollingTrigger)
             .toBuilder()
             .tasks(List.of(Fail.builder().id("fail").type(Fail.class.getName()).build()))
@@ -125,17 +125,62 @@ void pollingTriggerStopAfter() throws Exception {
         }
     }
 
+    @Test
+    void failedEvaluationTest() throws Exception {
+        // mock flow listener
+        FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
+        PollingTrigger pollingTrigger = createPollingTrigger(null)
+            .conditions(
+                List.of(
+                    VariableCondition.builder()
+                        .type(VariableCondition.class.getName())
+                        .expression("{{ trigger.date | date() < now() }}")
+                        .build()
+                ))
+            .build();
+        Flow flow = createPollingTriggerFlow(pollingTrigger);
+        doReturn(List.of(flow))
+            .when(flowListenersServiceSpy)
+            .flows();
+
+        CountDownLatch queueCount = new CountDownLatch(1);
+
+        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();
+                }
+            });
+
+            worker.run();
+            scheduler.run();
+
+            queueCount.await(10, TimeUnit.SECONDS);
+            // close the execution queue consumer
+            executionQueueStop.run();
+
+            assertThat(queueCount.getCount(), is(0L));
+            assertThat(last.get(), notNullValue());
+            assertThat(last.get().getState().getCurrent(), is(State.Type.FAILED));
+        }
+    }
+
     private Flow createPollingTriggerFlow(PollingTrigger pollingTrigger) {
         return createFlow(Collections.singletonList(pollingTrigger));
     }
 
-    private PollingTrigger createPollingTrigger(List<State.Type> stopAfter) {
+    private PollingTrigger.PollingTriggerBuilder<?, ?> createPollingTrigger(List<State.Type> stopAfter) {
         return PollingTrigger.builder()
             .id("polling-trigger")
             .type(PollingTrigger.class.getName())
             .duration(500L)
-            .stopAfter(stopAfter)
-            .build();
+            .stopAfter(stopAfter);
     }
 
     private AbstractScheduler scheduler(FlowListeners flowListenersServiceSpy) {
