Repository: kestra-io/kestra (java)

Issue:
Serialization issue when doing Property.of() on plugin-ldap repo
### Describe the issue

During migration to Property v2 for plugin-ldap ([PR](https://github.com/kestra-io/plugin-ldap/pull/23))
I encounter an issue when trying to use `Property.of(SearchScope.SUB)`. 

Here is the problematic line : https://github.com/kestra-io/plugin-ldap/blob/17eed04721bceed853b0c24ec8a892c296f9fdf7/src/main/java/io/kestra/plugin/ldap/Search.java#L103

Here is the stack trace associated : 
```
Caused by: java.lang.IllegalArgumentException: Cannot deserialize value of type `java.lang.String` from Object value (token `JsonToken.START_OBJECT`)
     at [Source: UNKNOWN; byte offset: #UNKNOWN]
    	at com.fasterxml.jackson.databind.ObjectMapper._convert(ObjectMapper.java:4636)
    	at com.fasterxml.jackson.databind.ObjectMapper.convertValue(ObjectMapper.java:4567)
    	at io.kestra.core.models.property.Property.of(Property.java:65)
    	at io.kestra.plugin.ldap.Search.$default$sub(Search.java:104)
    	at io.kestra.plugin.ldap.Search.<init>(Search.java:27)
    	at java.base/jdk.internal.reflect.DirectConstructorHandleAccessor.newInstance(DirectConstructorHandleAccessor.java:62)
    	at java.base/java.lang.reflect.Constructor.newInstanceWithCaller(Constructor.java:502)
    	at java.base/java.lang.reflect.Constructor.newInstance(Constructor.java:486)
    	at java.base/java.util.ServiceLoader$ProviderImpl.newInstance(ServiceLoader.java:789)
    	... 165 common frames omitted
    Caused by: com.fasterxml.jackson.databind.exc.MismatchedInputException: Cannot deserialize value of type `java.lang.String` from Object value (token `JsonToken.START_OBJECT`)
     at [Source: UNKNOWN; byte offset: #UNKNOWN]
    	at com.fasterxml.jackson.databind.exc.MismatchedInputException.from(MismatchedInputException.java:59)
    	at com.fasterxml.jackson.databind.DeserializationContext.reportInputMismatch(DeserializationContext.java:1767)
    	at com.fasterxml.jackson.databind.DeserializationContext.handleUnexpectedToken(DeserializationContext.java:1541)
    	at com.fasterxml.jackson.databind.DeserializationContext.handleUnexpectedToken(DeserializationContext.java:1446)
    	at com.fasterxml.jackson.databind.DeserializationContext.extractScalarFromObject(DeserializationContext.java:958)
    	at com.fasterxml.jackson.databind.deser.std.StdDeserializer._parseString(StdDeserializer.java:1441)
    	at com.fasterxml.jackson.databind.deser.std.StringDeserializer.deserialize(StringDeserializer.java:48)
    	at com.fasterxml.jackson.databind.deser.std.StringDeserializer.deserialize(StringDeserializer.java:11)
    	at com.fasterxml.jackson.databind.ObjectMapper._convert(ObjectMapper.java:4631)
```

### Environment

- Kestra Version: develop


Pull request description:
fix(core): Property.of() with arbitrary objects
Fixes #6109

Patch:
diff --git a/core/src/main/java/io/kestra/core/models/property/Property.java b/core/src/main/java/io/kestra/core/models/property/Property.java
index ec7745bb6..44050f78a 100644
--- a/core/src/main/java/io/kestra/core/models/property/Property.java
+++ b/core/src/main/java/io/kestra/core/models/property/Property.java
@@ -64,8 +64,18 @@ public class Property<T> {
                 throw new IllegalArgumentException(e);
             }
         } else {
-            expression = MAPPER.convertValue(value, String.class);
+            try {
+                expression = MAPPER.convertValue(value, String.class);
+            } catch (IllegalArgumentException e) {
+                // if it fails, try with writeValueAsString instead
+                try {
+                    expression = MAPPER.writeValueAsString(value);
+                } catch (JsonProcessingException e2) {
+                    throw new IllegalArgumentException(e2);
+                }
+            }
         }
+
         Property<V> p = new Property<>(expression);
         p.value = value;
         return p;
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 8eaa81dff..dc12e4e8c 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
@@ -24,9 +24,6 @@ public class Trigger extends TriggerContext implements HasUID {
     @Nullable
     private String executionId;
 
-    @Nullable
-    private State.Type executionCurrentState;
-
     @Nullable
     private Instant updatedDate;
 
@@ -40,7 +37,6 @@ public class Trigger extends TriggerContext implements HasUID {
     protected Trigger(TriggerBuilder<?, ?> b) {
         super(b);
         this.executionId = b.executionId;
-        this.executionCurrentState = b.executionCurrentState;
         this.updatedDate = b.updatedDate;
         this.evaluateRunningDate = b.evaluateRunningDate;
     }
@@ -145,7 +141,6 @@ public class Trigger extends TriggerContext implements HasUID {
             .date(trigger.getDate())
             .nextExecutionDate(trigger.getNextExecutionDate())
             .executionId(execution.getId())
-            .executionCurrentState(execution.getState().getCurrent())
             .updatedDate(Instant.now())
             .backfill(trigger.getBackfill())
             .stopAfter(trigger.getStopAfter())
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 f444739b1..658f071f0 100644
--- a/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
+++ b/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
@@ -73,6 +73,7 @@ public abstract class AbstractScheduler implements Scheduler, Service {
     private final PluginDefaultService pluginDefaultService;
     private final WorkerGroupService workerGroupService;
     private final LogService logService;
+    protected SchedulerExecutionStateInterface executionState;
 
     // must be volatile as it's updated by the flow listener thread and read by the scheduleExecutor thread
     private volatile Boolean isReady = false;
@@ -637,8 +638,10 @@ public abstract class AbstractScheduler implements Scheduler, Service {
             return true;
         }
 
-        // The execution is not yet started, we skip
-        if (lastTrigger.getExecutionCurrentState() == null) {
+        Optional<Execution> execution = executionState.findById(lastTrigger.getTenantId(), lastTrigger.getExecutionId());
+
+        // executionState hasn't received the execution, we skip
+        if (execution.isEmpty()) {
             if (lastTrigger.getUpdatedDate() != null) {
                 metricRegistry
                     .timer(MetricRegistry.SCHEDULER_EXECUTION_MISSING_DURATION, metricRegistry.tags(lastTrigger))
@@ -672,7 +675,7 @@ public abstract class AbstractScheduler implements Scheduler, Service {
                 Level.DEBUG,
                 "Execution '{}' is still '{}', updated at '{}'",
                 lastTrigger.getExecutionId(),
-                lastTrigger.getExecutionCurrentState(),
+                execution.get().getState().getCurrent(),
                 lastTrigger.getUpdatedDate()
             );
         }
diff --git a/core/src/main/java/io/kestra/core/schedulers/SchedulerExecutionState.java b/core/src/main/java/io/kestra/core/schedulers/SchedulerExecutionState.java
new file mode 100644
index 000000000..bf8acf5e7
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/schedulers/SchedulerExecutionState.java
@@ -0,0 +1,19 @@
+package io.kestra.core.schedulers;
+
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.repositories.ExecutionRepositoryInterface;
+
+import java.util.Optional;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+
+@Singleton
+public class SchedulerExecutionState implements SchedulerExecutionStateInterface {
+    @Inject
+    private ExecutionRepositoryInterface executionRepository;
+
+    @Override
+    public Optional<Execution> findById(String tenantId, String id) {
+        return executionRepository.findById(tenantId, id);
+    }
+}
diff --git a/core/src/main/java/io/kestra/core/schedulers/SchedulerExecutionStateInterface.java b/core/src/main/java/io/kestra/core/schedulers/SchedulerExecutionStateInterface.java
new file mode 100644
index 000000000..32652b0a9
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/schedulers/SchedulerExecutionStateInterface.java
@@ -0,0 +1,9 @@
+package io.kestra.core.schedulers;
+
+import io.kestra.core.models.executions.Execution;
+
+import java.util.Optional;
+
+public interface SchedulerExecutionStateInterface {
+    Optional<Execution> findById(String tenantId, String id);
+}
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 23bc39ab7..800776488 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcTriggerRepository.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcTriggerRepository.java
@@ -215,7 +215,6 @@ public abstract class AbstractJdbcTriggerRepository extends AbstractJdbcReposito
                     Trigger current = optionalTrigger.get();
                     current = current.toBuilder()
                         .executionId(trigger.getExecutionId())
-                        .executionCurrentState(trigger.getExecutionCurrentState())
                         .updatedDate(trigger.getUpdatedDate())
                         .build();
                     this.save(context, current);
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 aa9cea382..1c66692db 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcScheduler.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcScheduler.java
@@ -30,10 +30,10 @@ import java.util.function.BiConsumer;
 public class JdbcScheduler extends AbstractScheduler {
     private final QueueInterface<Execution> executionQueue;
     private final TriggerRepositoryInterface triggerRepository;
-    private final ConditionService conditionService;
 
     private final FlowRepositoryInterface flowRepository;
     private final JooqDSLContextWrapper dslContextWrapper;
+    private final ConditionService conditionService;
 
 
     @SuppressWarnings("unchecked")
@@ -47,6 +47,7 @@ public class JdbcScheduler extends AbstractScheduler {
         executionQueue = applicationContext.getBean(QueueInterface.class, Qualifiers.byName(QueueFactoryInterface.EXECUTION_NAMED));
         triggerRepository = applicationContext.getBean(AbstractJdbcTriggerRepository.class);
         triggerState = applicationContext.getBean(SchedulerTriggerStateInterface.class);
+        executionState = applicationContext.getBean(SchedulerExecutionState.class);
         conditionService = applicationContext.getBean(ConditionService.class);
         flowRepository = applicationContext.getBean(FlowRepositoryInterface.class);
         dslContextWrapper = applicationContext.getBean(JooqDSLContextWrapper.class);
@@ -74,14 +75,6 @@ public class JdbcScheduler extends AbstractScheduler {
                             .ifPresent(trigger -> {
                                 this.triggerState.update(trigger.resetExecution(execution.getState().getCurrent()));
                             });
-                    } else {
-                        // update execution state on each state change so the scheduler knows the execution is running
-                        triggerRepository
-                            .findByExecution(execution)
-                            .filter(trigger -> execution.getState().getCurrent() != trigger.getExecutionCurrentState())
-                            .ifPresent(trigger -> {
-                                ((JdbcSchedulerTriggerState) this.triggerState).updateExecution(Trigger.of(execution, trigger));
-                            });
                     }
                 }
             }
diff --git a/ui/src/components/admin/Triggers.vue b/ui/src/components/admin/Triggers.vue
index c8033695f..d46d76b27 100644
--- a/ui/src/components/admin/Triggers.vue
+++ b/ui/src/components/admin/Triggers.vue
@@ -127,16 +127,6 @@
                                 </router-link>
                             </template>
                         </el-table-column>
-
-                        <el-table-column v-if="visibleColumns.executionCurrentState" :label="$t('state')">
-                            <template #default="scope">
-                                <status
-                                    v-if="scope.row.executionCurrentState"
-                                    :status="scope.row.executionCurrentState"
-                                    size="small"
-                                />
-                            </template>
-                        </el-table-column>
                         <el-table-column v-if="visibleColumns.workerId" prop="workerId" :label="$t('workerId')">
                             <template #default="scope">
                                 <id
@@ -287,7 +277,6 @@
     import RefreshButton from "../layout/RefreshButton.vue";
     import DateAgo from "../layout/DateAgo.vue";
     import Id from "../Id.vue";
-    import Status from "../Status.vue";
     import {mapState} from "vuex";
     import SelectTableActions from "../../mixins/selectTableActions";
     import _merge from "lodash/merge";
@@ -302,7 +291,6 @@
             SearchField,
             NamespaceSelect,
             DateAgo,
-            Status,
             Id,
             LogsWrapper
         },


Test patch:
diff --git a/core/src/test/java/io/kestra/core/models/property/PropertyTest.java b/core/src/test/java/io/kestra/core/models/property/PropertyTest.java
index 479cc5ce7..148f2933c 100644
--- a/core/src/test/java/io/kestra/core/models/property/PropertyTest.java
+++ b/core/src/test/java/io/kestra/core/models/property/PropertyTest.java
@@ -6,6 +6,8 @@ import io.kestra.core.runners.RunContextFactory;
 import io.kestra.core.serializers.FileSerde;
 import io.kestra.core.storages.StorageInterface;
 import jakarta.inject.Inject;
+import lombok.Builder;
+import lombok.Getter;
 import org.junit.jupiter.api.Test;
 import org.slf4j.event.Level;
 import reactor.core.publisher.Flux;
@@ -222,4 +224,17 @@ class PropertyTest {
 
         assertThrows(IllegalVariableEvaluationException.class, () -> task.run(runContext));
     }
+
+    @Test
+    void of() {
+        var prop = Property.of(TestObj.builder().key("key").value("value").build());
+        assertThat(prop, notNullValue());
+    }
+
+    @Builder
+    @Getter
+    private static class TestObj {
+        private String key;
+        private String value;
+    }
 }
\ No newline at end of file
diff --git a/core/src/test/java/io/kestra/core/schedulers/SchedulerConditionTest.java b/core/src/test/java/io/kestra/core/schedulers/SchedulerConditionTest.java
index 110b62cbc..e38f654ae 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerConditionTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerConditionTest.java
@@ -18,14 +18,14 @@ import java.time.ZonedDateTime;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import static io.kestra.core.utils.Rethrow.throwConsumer;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.*;
 
 class SchedulerConditionTest extends AbstractSchedulerTest {
     @Inject
@@ -34,6 +34,9 @@ class SchedulerConditionTest extends AbstractSchedulerTest {
     @Inject
     protected SchedulerTriggerStateInterface triggerState;
 
+    @Inject
+    protected SchedulerExecutionStateInterface executionState;
+
     private static Flow createScheduleFlow() {
         Schedule schedule = Schedule.builder()
             .id("hourly")
@@ -59,6 +62,7 @@ class SchedulerConditionTest extends AbstractSchedulerTest {
     void schedule() throws Exception {
         // mock flow listeners
         FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
+        SchedulerExecutionStateInterface executionRepositorySpy = spy(this.executionState);
         CountDownLatch queueCount = new CountDownLatch(4);
 
         Flow flow = createScheduleFlow();
@@ -75,6 +79,11 @@ class SchedulerConditionTest extends AbstractSchedulerTest {
             .when(flowListenersServiceSpy)
             .flows();
 
+        // mock the backfill execution is ended
+        doAnswer(invocation -> Optional.of(Execution.builder().state(new State().withState(State.Type.SUCCESS)).build()))
+            .when(executionRepositorySpy)
+            .findById(any(), any());
+
         // scheduler
         try (AbstractScheduler scheduler = new JdbcScheduler(
             applicationContext,
@@ -95,7 +104,7 @@ class SchedulerConditionTest extends AbstractSchedulerTest {
             }));
 
             scheduler.run();
-            queueCount.await(30, TimeUnit.SECONDS);
+            queueCount.await(15, TimeUnit.SECONDS);
 
             receive.blockLast();
 
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 c4f798ffc..e5ff09812 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerPollingTriggerTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerPollingTriggerTest.java
@@ -40,6 +40,9 @@ public class SchedulerPollingTriggerTest extends AbstractSchedulerTest {
     @Inject
     private SchedulerTriggerStateInterface triggerState;
 
+    @Inject
+    private SchedulerExecutionState schedulerExecutionState;
+
     @Inject
     private FlowListeners flowListenersService;
 
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 1417b5043..6a3c8e071 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleTest.java
@@ -40,6 +40,9 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
     @Inject
     protected SchedulerTriggerStateInterface triggerState;
 
+    @Inject
+    protected SchedulerExecutionStateInterface executionState;
+
     @Inject
     @Named(QueueFactoryInterface.WORKERTASKLOG_NAMED)
     protected QueueInterface<LogEntry> logQueue;
@@ -67,7 +70,7 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
             .truncatedTo(ChronoUnit.HOURS);
     }
 
-    protected AbstractScheduler scheduler(FlowListeners flowListenersServiceSpy) {
+    protected AbstractScheduler scheduler(FlowListeners flowListenersServiceSpy, SchedulerExecutionStateInterface executionStateSpy) {
         return new JdbcScheduler(
             applicationContext,
             flowListenersServiceSpy
@@ -79,6 +82,7 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
     void schedule() throws Exception {
         // mock flow listeners
         FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
+        SchedulerExecutionStateInterface executionStateSpy = spy(this.executionState);
         CountDownLatch queueCount = new CountDownLatch(6);
         CountDownLatch invalidLogCount = new CountDownLatch(1);
         Set<String> date = new HashSet<>();
@@ -113,7 +117,7 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
         triggerState.create(trigger.toBuilder().triggerId("schedule-invalid").flowId(invalid.getId()).build());
 
         // scheduler
-        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy)) {
+        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy, executionStateSpy)) {
             // wait for execution
             Flux<Execution> receiveExecutions = TestsUtils.receive(executionQueue, throwConsumer(either -> {
                 Execution execution = either.getLeft();
@@ -173,7 +177,7 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
         triggerState.create(trigger);
 
         // scheduler
-        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy)) {
+        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy, executionState)) {
             scheduler.run();
 
             Await.until(() -> {
@@ -207,7 +211,7 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
         CountDownLatch queueCount = new CountDownLatch(1);
 
         // scheduler
-        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy)) {
+        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy, executionState)) {
             // wait for execution
             Flux<Execution> receive = TestsUtils.receive(executionQueue, either -> {
                 Execution execution = either.getLeft();
@@ -252,7 +256,7 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
         CountDownLatch queueCount = new CountDownLatch(1);
 
         // scheduler
-        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy)) {
+        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy, executionState)) {
             // wait for execution
             Flux<Execution> receive = TestsUtils.receive(executionQueue, either -> {
                 Execution execution = either.getLeft();
@@ -296,7 +300,7 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
         triggerState.create(lastTrigger);
 
         // scheduler
-        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy)) {
+        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy, executionState)) {
             scheduler.run();
 
             Await.until(() -> scheduler.isReady(), Duration.ofMillis(100), Duration.ofSeconds(5));
@@ -327,7 +331,7 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
             .build();
 
         // scheduler
-        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy)) {
+        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy, executionState)) {
             scheduler.run();
 
             Await.until(() -> {
@@ -392,7 +396,7 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
         triggerState.create(trigger);
 
         // scheduler
-        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy)) {
+        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy, executionState)) {
             scheduler.run();
 
             // Wait 3s to see if things happen
@@ -430,7 +434,7 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
         CountDownLatch queueCount = new CountDownLatch(2);
 
         // scheduler
-        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy)) {
+        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy, executionState)) {
             // wait for execution
             Flux<Execution> receive = TestsUtils.receive(executionQueue, throwConsumer(either -> {
                 Execution execution = either.getLeft();
@@ -490,7 +494,7 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
         CountDownLatch queueCount = new CountDownLatch(1);
 
         // scheduler
-        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy)) {
+        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy, executionState)) {
             // wait for execution
             Flux<Execution> receive = TestsUtils.receive(executionQueue, either -> {
                 Execution execution = either.getLeft();
diff --git a/core/src/test/java/io/kestra/core/schedulers/SchedulerThreadTest.java b/core/src/test/java/io/kestra/core/schedulers/SchedulerThreadTest.java
index 872fb21d5..36692a2eb 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerThreadTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerThreadTest.java
@@ -19,21 +19,21 @@ import reactor.core.publisher.Flux;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import static io.kestra.core.utils.Rethrow.throwConsumer;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.*;
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.*;
 
 public class SchedulerThreadTest extends AbstractSchedulerTest {
     @Inject
     protected FlowListeners flowListenersService;
 
     @Inject
-    protected SchedulerTriggerStateInterface triggerState;
+    protected SchedulerExecutionStateInterface executionState;
 
     @Test
     void thread() throws Exception {
@@ -54,12 +54,17 @@ public class SchedulerThreadTest extends AbstractSchedulerTest {
 
         // mock flow listeners
         FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
-
+        SchedulerExecutionStateInterface schedulerExecutionStateSpy = spy(this.executionState);
 
         doReturn(Collections.singletonList(flow))
             .when(flowListenersServiceSpy)
             .flows();
 
+        // mock the backfill execution is ended
+        doAnswer(invocation -> Optional.of(Execution.builder().state(new State().withState(State.Type.SUCCESS)).build()))
+            .when(schedulerExecutionStateSpy)
+            .findById(any(), any());
+
         // scheduler
         try (
             AbstractScheduler scheduler = new JdbcScheduler(
diff --git a/jdbc-h2/src/test/java/io/kestra/schedulers/h2/H2SchedulerScheduleTest.java b/jdbc-h2/src/test/java/io/kestra/schedulers/h2/H2SchedulerScheduleTest.java
index 059860c6c..2d6211ff5 100644
--- a/jdbc-h2/src/test/java/io/kestra/schedulers/h2/H2SchedulerScheduleTest.java
+++ b/jdbc-h2/src/test/java/io/kestra/schedulers/h2/H2SchedulerScheduleTest.java
@@ -2,12 +2,13 @@ package io.kestra.schedulers.h2;
 
 import io.kestra.core.runners.FlowListeners;
 import io.kestra.core.schedulers.AbstractScheduler;
+import io.kestra.core.schedulers.SchedulerExecutionStateInterface;
 import io.kestra.core.schedulers.SchedulerScheduleTest;
 import io.kestra.jdbc.runner.JdbcScheduler;
 
 class H2SchedulerScheduleTest extends SchedulerScheduleTest {
     @Override
-    protected AbstractScheduler scheduler(FlowListeners flowListenersServiceSpy) {
+    protected AbstractScheduler scheduler(FlowListeners flowListenersServiceSpy, SchedulerExecutionStateInterface executionStateSpy) {
         return new JdbcScheduler(
             applicationContext,
             flowListenersServiceSpy
diff --git a/jdbc-mysql/src/test/java/io/kestra/schedulers/mysql/MysqlSchedulerScheduleTest.java b/jdbc-mysql/src/test/java/io/kestra/schedulers/mysql/MysqlSchedulerScheduleTest.java
index 70563f348..6e017199e 100644
--- a/jdbc-mysql/src/test/java/io/kestra/schedulers/mysql/MysqlSchedulerScheduleTest.java
+++ b/jdbc-mysql/src/test/java/io/kestra/schedulers/mysql/MysqlSchedulerScheduleTest.java
@@ -2,12 +2,13 @@ package io.kestra.schedulers.mysql;
 
 import io.kestra.core.runners.FlowListeners;
 import io.kestra.core.schedulers.AbstractScheduler;
+import io.kestra.core.schedulers.SchedulerExecutionStateInterface;
 import io.kestra.core.schedulers.SchedulerScheduleTest;
 import io.kestra.jdbc.runner.JdbcScheduler;
 
 class MysqlSchedulerScheduleTest extends SchedulerScheduleTest {
     @Override
-    protected AbstractScheduler scheduler(FlowListeners flowListenersServiceSpy) {
+    protected AbstractScheduler scheduler(FlowListeners flowListenersServiceSpy, SchedulerExecutionStateInterface executionStateSpy) {
         return new JdbcScheduler(
             applicationContext,
             flowListenersServiceSpy
diff --git a/jdbc-postgres/src/test/java/io/kestra/schedulers/postgres/PostgresSchedulerScheduleTest.java b/jdbc-postgres/src/test/java/io/kestra/schedulers/postgres/PostgresSchedulerScheduleTest.java
index 14217d717..06ad6f687 100644
--- a/jdbc-postgres/src/test/java/io/kestra/schedulers/postgres/PostgresSchedulerScheduleTest.java
+++ b/jdbc-postgres/src/test/java/io/kestra/schedulers/postgres/PostgresSchedulerScheduleTest.java
@@ -2,12 +2,13 @@ package io.kestra.schedulers.postgres;
 
 import io.kestra.core.runners.FlowListeners;
 import io.kestra.core.schedulers.AbstractScheduler;
+import io.kestra.core.schedulers.SchedulerExecutionStateInterface;
 import io.kestra.core.schedulers.SchedulerScheduleTest;
 import io.kestra.jdbc.runner.JdbcScheduler;
 
 class PostgresSchedulerScheduleTest extends SchedulerScheduleTest {
     @Override
-    protected AbstractScheduler scheduler(FlowListeners flowListenersServiceSpy) {
+    protected AbstractScheduler scheduler(FlowListeners flowListenersServiceSpy, SchedulerExecutionStateInterface executionStateSpy) {
         return new JdbcScheduler(
             applicationContext,
             flowListenersServiceSpy