Repository: kestra-io/kestra (java)

Issue:
Add handling for trailing slash in the KESTRA_URL configuration to avoid missing or double slash
### Describe the issue

When I executed a flow using the API, I got a response with a URL to Kestra that contained two slashes when only one was needed.

Flow:
```yaml
id: hello_world
namespace: company.team

inputs:
  - id: greeting
    type: STRING
    defaults: hey

tasks:
  - id: hello
    type: io.kestra.plugin.core.log.Log
    message: "{{ inputs.greeting }}"
```

API Request made in the terminal:

```sh
curl -X POST http://localhost:8084/api/v1/executions/company.team/hello_world
```

Response given:

```json
{
    "id": "6gTABm82mx5li7tHkgqyNU",
    "namespace": "company.team",
    "flowId": "hello_world",
    "flowRevision": 10,
    "inputs": {
        "greeting": "hey"
    },
    "labels": [
        {
            "key": "system.correlationId",
            "value": "6gTABm82mx5li7tHkgqyNU"
        }
    ],
    "state": {
        "current": "CREATED",
        "histories": [
            {
                "state": "CREATED",
                "date": "2024-11-21T16:35:36.815338221Z"
            }
        ],
        "duration": 0.012298625,
        "startDate": "2024-11-21T16:35:36.815338221Z"
    },
    "originalId": "6gTABm82mx5li7tHkgqyNU",
    "deleted": false,
    "metadata": {
        "attemptNumber": 1,
        "originalCreatedDate": "2024-11-21T16:35:36.815378387Z"
    },
    "url": "http://localhost:8084//ui/executions/company.team/hello_world/6gTABm82mx5li7tHkgqyNU"
}
```

Specifically the bottom line `url` has two `/` after the server URL. It should be:
`"url": "http://localhost:8084/ui/executions/company.team/hello_world/6gTABm82mx5li7tHkgqyNU"`

Should be a quick fix, and doesn't stop it working - just a minor bug.

### Environment

- Kestra Version: develop OSS 78f5ff2


Pull request description:
fix(core): Handling for trailing slash in the KESTRA_URL configuration
<!-- Thanks for submitting a Pull Request to Kestra. To help us review your contribution, please follow the guidelines below:

- Make sure that your commits follow the [conventional commits](https://www.conventionalcommits.org/en/v1.0.0/) specification e.g. `feat(ui): add a new navigation menu item` or `fix(core): fix a bug in the core model` or `docs: update the README.md`. This will help us automatically generate the changelog.
- The title should briefly summarize the proposed changes.
- Provide a short overview of the change and the value it adds.
- Share a flow example to help the reviewer understand and QA the change.
- Use "closes" to automatically close an issue. For example, `closes #1234` will close issue #1234. -->

closes #6041
### What changes are being made and why?

<!-- Please include a brief summary of the changes included in this PR e.g. closes #1234. -->

---

This fix covers the cases when the trailing slash is added mistakenly to the configuration. It doesn't validate the whole URL, this could be considered in the future. 

This will be the behavior after this is merged.

Input: "http://localhost:8081/" -> Output: "http://localhost:8081"
Input: "http://localhost:8081" -> Output: "http://localhost:8081"
Input: "http://localhost:8081//" -> Output: "http://localhost:8081/" (because only the last slash is removed).


### How the changes have been QAed?

<!-- Include example code that shows how this PR has been QAed. The code should present a complete yet easily reproducible flow.

```yaml
# Your example flow code here
```

Note that this is not a replacement for unit tests but rather a way to demonstrate how the changes work in a real-life scenario, as the end-user would experience them.

Remove this section if this change applies to all flows or to the documentation only. -->

---

The tests in the `ExecutionControllerTest` are passing when the trailing slash is added to the kestra.url in the application-test.yml.


<!--If there are any setup requirements like API keys or trial accounts, kindly include brief bullet-points-description outlining the setup process below.

- [External System Documentation](URL)
- Steps to set up the necessary resources

If there are no setup requirements, you can remove this section.

Thank you for your contribution. ❤️  -->


Patch:
diff --git a/core/src/main/java/io/kestra/core/runners/Executor.java b/core/src/main/java/io/kestra/core/runners/Executor.java
index 22361abb4..2cf698db4 100644
--- a/core/src/main/java/io/kestra/core/runners/Executor.java
+++ b/core/src/main/java/io/kestra/core/runners/Executor.java
@@ -4,6 +4,7 @@ import com.fasterxml.jackson.annotation.JsonIgnore;
 import io.kestra.core.models.executions.*;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.FlowWithException;
+import io.kestra.core.models.flows.FlowWithSource;
 import io.kestra.core.models.flows.State;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
@@ -20,7 +21,7 @@ public class Executor {
     private Long offset;
     @JsonIgnore
     private boolean executionUpdated = false;
-    private Flow flow;
+    private FlowWithSource flow;
     private final List<TaskRun> nexts = new ArrayList<>();
     private final List<WorkerTask> workerTasks = new ArrayList<>();
     private final List<ExecutionDelay> executionDelays = new ArrayList<>();
@@ -88,7 +89,7 @@ public class Executor {
         return !(this.getException() != null || this.getFlow() == null || this.getFlow() instanceof FlowWithException || this.getFlow().getTasks() == null || this.getExecution().isDeleted() || this.getExecution().getState().isPaused());
     }
 
-    public Executor withFlow(Flow flow) {
+    public Executor withFlow(FlowWithSource flow) {
         this.flow = flow;
 
         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 342abc869..6d6e3ded8 100644
--- a/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
+++ b/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
@@ -1080,12 +1080,4 @@ public abstract class AbstractScheduler implements Scheduler, Service {
     public ServiceState getState() {
         return state.get();
     }
-
-    protected Trigger resetExecution(FlowWithSource flow, Execution execution, Trigger trigger) {
-        Flow flowWithDefaults = pluginDefaultService.injectDefaults(flow, execution);
-        RunContext runContext = runContextFactory.of(flowWithDefaults, flowWithDefaults.findTriggerByTriggerId(trigger.getTriggerId()));
-        ConditionContext conditionContext = conditionService.conditionContext(runContext, flowWithDefaults, null);
-
-        return trigger.resetExecution(flowWithDefaults, execution, conditionContext);
-    }
 }
diff --git a/core/src/main/java/io/kestra/core/services/ExecutionService.java b/core/src/main/java/io/kestra/core/services/ExecutionService.java
index df68cb890..bcedebb42 100644
--- a/core/src/main/java/io/kestra/core/services/ExecutionService.java
+++ b/core/src/main/java/io/kestra/core/services/ExecutionService.java
@@ -4,12 +4,14 @@ import io.kestra.core.events.CrudEvent;
 import io.kestra.core.events.CrudEventType;
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.models.Label;
+import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.ExecutionKilled;
 import io.kestra.core.models.executions.ExecutionKilledExecution;
 import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.models.executions.TaskRunAttempt;
 import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.flows.FlowWithSource;
 import io.kestra.core.models.flows.State;
 import io.kestra.core.models.flows.input.InputAndValue;
 import io.kestra.core.models.hierarchies.AbstractGraphTask;
@@ -17,6 +19,7 @@ import io.kestra.core.models.hierarchies.GraphCluster;
 import io.kestra.core.models.tasks.ResolvedTask;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.tasks.retrys.AbstractRetry;
+import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
 import io.kestra.core.repositories.ExecutionRepositoryInterface;
@@ -24,6 +27,8 @@ import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.repositories.LogRepositoryInterface;
 import io.kestra.core.repositories.MetricRepositoryInterface;
 import io.kestra.core.runners.FlowInputOutput;
+import io.kestra.core.runners.RunContext;
+import io.kestra.core.runners.RunContextFactory;
 import io.kestra.core.storages.StorageContext;
 import io.kestra.core.storages.StorageInterface;
 import io.kestra.core.utils.GraphUtils;
@@ -94,6 +99,12 @@ public class ExecutionService {
     @Inject
     private ConditionService conditionService;
 
+    @Inject
+    private RunContextFactory runContextFactory;
+
+    @Inject
+    private PluginDefaultService pluginDefaultService;
+
     public Execution getExecutionIfPause(final String tenant, final @NotNull String executionId, boolean withACL) {
         Execution execution = getExecution(tenant, executionId, withACL);
 
@@ -858,4 +869,19 @@ public class ExecutionService {
             .map(ResolvedTask::of)
             .toList();
     }
+
+    /**
+     * Reset a trigger after an execution was terminated
+     */
+    public Trigger resetExecution(FlowWithSource flow, Execution execution, Trigger trigger) {
+        if (!execution.getState().isTerminated()) {
+            throw new IllegalArgumentException("Only terminated executions can be reset.");
+        }
+
+        FlowWithSource flowWithDefaults = pluginDefaultService.injectDefaults(flow, execution);
+        RunContext runContext = runContextFactory.of(flowWithDefaults, flowWithDefaults.findTriggerByTriggerId(trigger.getTriggerId()));
+        ConditionContext conditionContext = conditionService.conditionContext(runContext, flowWithDefaults, null);
+
+        return trigger.resetExecution(flowWithDefaults, execution, conditionContext);
+    }
 }
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
index e18bd7332..4b55b4073 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
@@ -21,9 +21,11 @@ import io.kestra.core.queues.QueueException;
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
 import io.kestra.core.repositories.FlowRepositoryInterface;
+import io.kestra.core.repositories.TriggerRepositoryInterface;
 import io.kestra.core.runners.Executor;
 import io.kestra.core.runners.ExecutorService;
 import io.kestra.core.runners.*;
+import io.kestra.core.schedulers.SchedulerTriggerStateInterface;
 import io.kestra.core.server.ClusterEvent;
 import io.kestra.core.server.Service;
 import io.kestra.core.server.ServiceStateChangeEvent;
@@ -174,6 +176,12 @@ public class JdbcExecutor implements ExecutorInterface, Service {
     @Inject
     private SLAService slaService;
 
+    @Inject
+    private TriggerRepositoryInterface triggerRepository;
+
+    @Inject
+    private SchedulerTriggerStateInterface triggerState;
+
     @Value("${kestra.jdbc.executor.thread-count:0}")
     private int threadCount;
 
@@ -447,7 +455,7 @@ public class JdbcExecutor implements ExecutorInterface, Service {
                 () -> {
                     try {
 
-                        final Flow flow = transform(this.flowRepository.findByExecutionWithSource(execution), execution);
+                        final FlowWithSource flow = transform(this.flowRepository.findByExecutionWithSource(execution), execution);
                         Executor executor = new Executor(execution, null).withFlow(flow);
 
                         // schedule it for later if needed
@@ -1008,6 +1016,16 @@ public class JdbcExecutor implements ExecutorInterface, Service {
                     );
                 }
 
+                // purge the trigger: reset scheduler trigger at end
+                if (execution.getTrigger() != null) {
+                    FlowWithSource flow = executor.getFlow();
+                    triggerRepository
+                        .findByExecution(execution)
+                        .ifPresent(trigger -> {
+                            this.triggerState.update(executionService.resetExecution(flow, execution, trigger));
+                        });
+                }
+
                 // Purge the workerTaskResultQueue and the workerJobQueue
                 // IMPORTANT: this is safe as only the executor is listening to WorkerTaskResult,
                 // and we are sure at this stage that all WorkerJob has been listened and processed by the Worker.
@@ -1036,7 +1054,7 @@ public class JdbcExecutor implements ExecutorInterface, Service {
         }
     }
 
-    private Flow transform(FlowWithSource flow, Execution execution) {
+    private FlowWithSource transform(FlowWithSource flow, Execution execution) {
         if (templateExecutorInterface.isPresent()) {
             try {
                 flow = Template.injectTemplate(
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 9d066a1fd..0607532f0 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcScheduler.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcScheduler.java
@@ -1,13 +1,9 @@
 package io.kestra.jdbc.runner;
 
-import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.FlowWithSource;
 import io.kestra.core.models.triggers.Trigger;
-import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.repositories.TriggerRepositoryInterface;
 import io.kestra.core.schedulers.*;
-import io.kestra.core.services.ConditionService;
-import io.kestra.core.services.ExecutionService;
 import io.kestra.core.services.FlowListenersInterface;
 import io.kestra.core.services.FlowService;
 import io.kestra.core.utils.ListUtils;
@@ -27,9 +23,7 @@ import java.util.function.BiConsumer;
 @Slf4j
 public class JdbcScheduler extends AbstractScheduler {
     private final TriggerRepositoryInterface triggerRepository;
-    private final FlowRepositoryInterface flowRepository;
     private final JooqDSLContextWrapper dslContextWrapper;
-    private final ExecutionService executionService;
 
 
     @Inject
@@ -42,8 +36,6 @@ public class JdbcScheduler extends AbstractScheduler {
         triggerRepository = applicationContext.getBean(AbstractJdbcTriggerRepository.class);
         triggerState = applicationContext.getBean(SchedulerTriggerStateInterface.class);
         executionState = applicationContext.getBean(SchedulerExecutionState.class);
-        executionService = applicationContext.getBean(ExecutionService.class);
-        flowRepository = applicationContext.getBean(FlowRepositoryInterface.class);
         dslContextWrapper = applicationContext.getBean(JooqDSLContextWrapper.class);
     }
 
@@ -51,29 +43,6 @@ public class JdbcScheduler extends AbstractScheduler {
     public void run() {
         super.run();
 
-        this.receiveCancellations.addFirst(executionQueue.receive(
-            Scheduler.class,
-            either -> {
-                if (either.isRight()) {
-                    log.error("Unable to deserialize an execution: {}", either.getRight().getMessage());
-                    return;
-                }
-
-                Execution execution = either.getLeft();
-                if (execution.getTrigger() != null) {
-                    var flow = flowRepository.findByIdWithSource(execution.getTenantId(), execution.getNamespace(), execution.getFlowId()).orElse(null);
-                    if (execution.isDeleted() || executionService.isTerminated(flow, execution)) {
-                        // reset scheduler trigger at end
-                        triggerRepository
-                            .findByExecution(execution)
-                            .ifPresent(trigger -> {
-                                this.triggerState.update(resetExecution(flow, execution, trigger));
-                            });
-                    }
-                }
-            }
-        ));
-
         // remove trigger on flow update
         this.flowListeners.listen((flow, previous) -> {
             if (flow.isDeleted()) {
diff --git a/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java b/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
index 5846917d5..30dbae99c 100644
--- a/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
+++ b/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
@@ -672,7 +672,8 @@ public class ExecutionController {
     }
 
     private URI executionUrl(Execution execution) {
-        return URI.create(kestraUrl.orElse("") + "/ui" + (execution.getTenantId() != null ? "/" + execution.getTenantId(): "")
+        String baseUrl = kestraUrl.map(url -> url.endsWith("/") ? url.substring(0, url.length() - 1) : url).orElse("");
+        return URI.create(baseUrl + "/ui" + (execution.getTenantId() != null ? "/" + execution.getTenantId(): "")
             + "/executions/"
             + execution.getNamespace() + "/"
             + execution.getFlowId() + "/"


Test patch:
diff --git a/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java b/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java
index 6083564f5..388258908 100644
--- a/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java
@@ -11,9 +11,12 @@ import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.WorkerGroup;
 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.queues.QueueException;
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
+import io.kestra.core.services.ExecutionService;
 import io.kestra.plugin.core.debug.Return;
 import io.kestra.core.utils.IdUtils;
 import io.kestra.plugin.core.flow.Sleep;
@@ -39,6 +42,12 @@ abstract public class AbstractSchedulerTest {
     @Named(QueueFactoryInterface.EXECUTION_NAMED)
     protected QueueInterface<Execution> executionQueue;
 
+    @Inject
+    protected Optional<SchedulerTriggerStateInterface> triggerState;
+
+    @Inject
+    protected ExecutionService executionService;
+
     public static Flow createThreadFlow() {
         return createThreadFlow(null);
     }
@@ -113,6 +122,22 @@ abstract public class AbstractSchedulerTest {
             .build();
     }
 
+    protected void terminateExecution(Execution execution, Trigger trigger, FlowWithSource flow) throws QueueException {
+        terminateExecution(execution, State.Type.SUCCESS, trigger, flow);
+    }
+
+    protected void terminateExecution(Execution execution, State.Type newState, Trigger trigger, FlowWithSource flow) throws QueueException {
+        if (triggerState.isEmpty()) {
+            throw new IllegalStateException("No triggerState available in the bean factory");
+        }
+
+        Execution terminated = execution.withState(newState);
+        executionQueue.emit(terminated);
+        triggerState.get().findLast(trigger)
+            .ifPresent(t -> triggerState.get().update(executionService.resetExecution(flow, terminated, t)));
+    }
+
+
     protected static int COUNTER = 0;
 
     @SuperBuilder
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 d450275f0..03f4034f6 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerConditionTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerConditionTest.java
@@ -26,6 +26,7 @@ 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.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.*;
 
 class SchedulerConditionTest extends AbstractSchedulerTest {
@@ -41,6 +42,7 @@ class SchedulerConditionTest extends AbstractSchedulerTest {
     @Inject
     protected FlowRepositoryInterface flowRepository;
 
+
     private static Flow createScheduleFlow() {
         Schedule schedule = Schedule.builder()
             .id("hourly")
@@ -72,13 +74,13 @@ class SchedulerConditionTest extends AbstractSchedulerTest {
         Flow flow = createScheduleFlow();
         flowRepository.create(flow, flow.generateSource(), flow);
 
-        triggerState.create(Trigger.builder()
+        Trigger trigger = Trigger.builder()
             .namespace(flow.getNamespace())
             .flowId(flow.getId())
             .triggerId("hourly")
             .date(ZonedDateTime.parse("2021-09-06T02:00:00+01:00[Europe/Paris]"))
-            .build()
-        );
+            .build();
+        triggerState.create(trigger);
 
         doReturn(Collections.singletonList(flow))
             .when(flowListenersServiceSpy)
@@ -98,7 +100,7 @@ class SchedulerConditionTest extends AbstractSchedulerTest {
             Flux<Execution> receive = TestsUtils.receive(executionQueue, throwConsumer(either -> {
                 Execution execution = either.getLeft();
                 if (execution.getState().getCurrent() == State.Type.CREATED) {
-                    executionQueue.emit(execution.withState(State.Type.SUCCESS));
+                    terminateExecution(execution, trigger, flow.withSource(flow.generateSource()));
 
                     queueCount.countDown();
                     if (queueCount.getCount() == 0) {
@@ -109,11 +111,8 @@ class SchedulerConditionTest extends AbstractSchedulerTest {
             }));
 
             scheduler.run();
-            queueCount.await(15, TimeUnit.SECONDS);
-
+            assertTrue(queueCount.await(15, TimeUnit.SECONDS));
             receive.blockLast();
-
-            assertThat(queueCount.getCount(), is(0L));
         }
     }
 }
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 6af3e5c5f..1a94d6fa1 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerPollingTriggerTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerPollingTriggerTest.java
@@ -114,7 +114,7 @@ public class SchedulerPollingTriggerTest extends AbstractSchedulerTest {
                     queueCount.countDown();
 
                     if (execution.getLeft().getState().getCurrent() == State.Type.CREATED) {
-                        executionQueue.emit(execution.getLeft().withState(State.Type.FAILED));
+                        terminateExecution(execution.getLeft(), State.Type.FAILED, Trigger.of(flow, pollingTrigger), flow.withSource(flow.generateSource()));
                     }
                 }
             }));
diff --git a/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleOnDatesTest.java b/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleOnDatesTest.java
index b7aa4694c..5c60bb1b4 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleOnDatesTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleOnDatesTest.java
@@ -107,7 +107,7 @@ public class SchedulerScheduleOnDatesTest extends AbstractSchedulerTest {
                 executionId.add(execution.getId());
 
                 if (execution.getState().getCurrent() == State.Type.CREATED) {
-                    executionQueue.emit(execution.withState(State.Type.SUCCESS));
+                    terminateExecution(execution, trigger, flow.withSource(flow.generateSource()));
                 }
                 assertThat(execution.getFlowId(), is(flow.getId()));
                 queueCount.countDown();
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 268ed2f2c..00486d7ef 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleTest.java
@@ -135,7 +135,7 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
                 executionId.add(execution.getId());
 
                 if (execution.getState().getCurrent() == State.Type.CREATED) {
-                    executionQueue.emit(execution.withState(State.Type.SUCCESS));
+                    terminateExecution(execution, trigger, flow.withSource(flow.generateSource()));
                 }
                 assertThat(execution.getFlowId(), is(flow.getId()));
                 queueCount.countDown();
@@ -451,7 +451,7 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
                 assertThat(execution.getFlowId(), is(flow.getId()));
 
                 if (execution.getState().getCurrent() == State.Type.CREATED) {
-                    executionQueue.emit(execution.withState(State.Type.SUCCESS));
+                    terminateExecution(execution, lastTrigger, flow.withSource(flow.generateSource()));
                 }
                 queueCount.countDown();
             }));
@@ -636,20 +636,17 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
                 assertThat(execution.getFlowId(), is(flow.getId()));
 
                 if (execution.getState().getCurrent() == State.Type.CREATED) {
-                    Thread.sleep(10000);
-                    executionQueue.emit(execution.withState(State.Type.SUCCESS)
-                        .toBuilder()
-                        .taskRunList(List.of(TaskRun.builder()
-                            .id("test")
-                            .executionId(execution.getId())
-                            .state(State.of(State.Type.SUCCESS,
-                                List.of(new State.History(
-                                    State.Type.SUCCESS,
-                                    lastTrigger.getNextExecutionDate().plusMinutes(3).toInstant()
-                                ))))
-                            .build()))
-                        .build()
-                    );
+                    Thread.sleep(11000);
+                    Execution terminated = execution.withTaskRunList(List.of(TaskRun.builder()
+                        .id("test")
+                        .executionId(execution.getId())
+                        .state(State.of(State.Type.SUCCESS,
+                            List.of(new State.History(
+                                State.Type.SUCCESS,
+                                lastTrigger.getNextExecutionDate().plusMinutes(3).toInstant()
+                            ))))
+                        .build()));
+                    terminateExecution(terminated, lastTrigger, flow.withSource(flow.generateSource()));
                 }
                 queueCount.countDown();
             }));
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 800ee8fe2..532479bb6 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerThreadTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerThreadTest.java
@@ -4,6 +4,7 @@ import io.kestra.core.models.Label;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.State;
+import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.runners.FlowListeners;
 import io.kestra.core.runners.TestMethodScopedWorker;
@@ -49,7 +50,7 @@ public class SchedulerThreadTest extends AbstractSchedulerTest {
             assertThat(execution.getFlowId(), is(flow.getId()));
 
             if (execution.getState().getCurrent() != State.Type.SUCCESS) {
-                executionQueue.emit(execution.withState(State.Type.SUCCESS));
+                terminateExecution(execution, Trigger.of(flow, flow.getTriggers().getFirst()), flow.withSource(flow.generateSource()));
                 queueCount.countDown();
             }
         }));