Repository: kestra-io/kestra
File: core/src/main/java/io/kestra/core/services/ExecutionService.java (Java)

Code before:
                .forEach(task -> newTaskRuns.removeIf(taskRun -> taskRun.getTaskId().equals(task.getId())));

            // We need to remove afterExecution tasks
            ListUtils.emptyOnNull(flow.getAfterExecution())
                .forEach(task -> newTaskRuns.removeIf(taskRun -> taskRun.getTaskId().equals(task.getId())));

            return newExecution.withTaskRunList(newTaskRuns);
        } else {
            return newExecution;
        }
    }

    public Execution markAs(final Execution execution, FlowInterface flow, String taskRunId, State.Type newState) throws Exception {
        return this.markAs(execution, flow, taskRunId, newState, null, null);
    }

    @SuppressWarnings("deprecation")
    private Execution markAs(final Execution execution, FlowInterface flow, String taskRunId, State.Type newState, @Nullable Map<String, Object> onResumeInputs, @Nullable Pause.Resumed resumed) throws Exception {
        Set<String> taskRunToRestart = this.taskRunToRestart(
            execution,
            taskRun -> taskRun.getId().equals(taskRunId)
        );

        Execution newExecution = execution.withMetadata(execution.getMetadata().nextAttempt());

        final FlowWithSource flowWithSource = pluginDefaultService.injectVersionDefaults(flow, false );

        for (String s : taskRunToRestart) {
            TaskRun originalTaskRun = newExecution.findTaskRunByTaskRunId(s);
            Task task = flowWithSource.findTaskByTaskId(originalTaskRun.getTaskId());
            boolean isFlowable = task.isFlowable();

            if (!isFlowable || s.equals(taskRunId)) {
                TaskRun newTaskRun;

                State.Type targetState = newState;
                if (task instanceof Pause pauseTask) {
                    State.Type terminalState = newState == State.Type.RUNNING ? State.Type.SUCCESS : newState;
                    Pause.Resumed _resumed = resumed != null ? resumed : Pause.Resumed.now(terminalState);
                    Variables variables = variablesService.of(StorageContext.forTask(originalTaskRun), pauseTask.generateOutputs(onResumeInputs, _resumed));
                    newTaskRun = originalTaskRun.withOutputs(variables);

                    // if it's a Pause task with no subtask, we terminate the task
                    if (ListUtils.isEmpty(pauseTask.getTasks()) && ListUtils.isEmpty(pauseTask.getErrors()) && ListUtils.isEmpty(pauseTask.getFinally())) {
                        if (newState == State.Type.RUNNING) {
                            targetState = State.Type.SUCCESS;
                        } else if (newState == State.Type.KILLING) {
                            targetState = State.Type.KILLED;
                        }
                    } else {
                        // we should set the state to RUNNING so that subtasks are executed

Review comment (suggestion):
```suggestion
        final FlowWithSource flowWithSource = pluginDefaultService.injectVersionDefaults(flow, false);
```

Code after:
                .forEach(task -> newTaskRuns.removeIf(taskRun -> taskRun.getTaskId().equals(task.getId())));

            // We need to remove afterExecution tasks
            ListUtils.emptyOnNull(flow.getAfterExecution())
                .forEach(task -> newTaskRuns.removeIf(taskRun -> taskRun.getTaskId().equals(task.getId())));

            return newExecution.withTaskRunList(newTaskRuns);
        } else {
            return newExecution;
        }
    }

    public Execution markAs(final Execution execution, FlowInterface flow, String taskRunId, State.Type newState) throws Exception {
        return this.markAs(execution, flow, taskRunId, newState, null, null);
    }

    @SuppressWarnings("deprecation")
    private Execution markAs(final Execution execution, FlowInterface flow, String taskRunId, State.Type newState, @Nullable Map<String, Object> onResumeInputs, @Nullable Pause.Resumed resumed) throws Exception {
        Set<String> taskRunToRestart = this.taskRunToRestart(
            execution,
            taskRun -> taskRun.getId().equals(taskRunId)
        );

        Execution newExecution = execution.withMetadata(execution.getMetadata().nextAttempt());

        final FlowWithSource flowWithSource = pluginDefaultService.injectVersionDefaults(flow, false);

        for (String s : taskRunToRestart) {
            TaskRun originalTaskRun = newExecution.findTaskRunByTaskRunId(s);
            Task task = flowWithSource.findTaskByTaskId(originalTaskRun.getTaskId());
            boolean isFlowable = task.isFlowable();

            if (!isFlowable || s.equals(taskRunId)) {
                TaskRun newTaskRun;

                State.Type targetState = newState;
                if (task instanceof Pause pauseTask) {
                    State.Type terminalState = newState == State.Type.RUNNING ? State.Type.SUCCESS : newState;
                    Pause.Resumed _resumed = resumed != null ? resumed : Pause.Resumed.now(terminalState);
                    Variables variables = variablesService.of(StorageContext.forTask(originalTaskRun), pauseTask.generateOutputs(onResumeInputs, _resumed));
                    newTaskRun = originalTaskRun.withOutputs(variables);

                    // if it's a Pause task with no subtask, we terminate the task
                    if (ListUtils.isEmpty(pauseTask.getTasks()) && ListUtils.isEmpty(pauseTask.getErrors()) && ListUtils.isEmpty(pauseTask.getFinally())) {
                        if (newState == State.Type.RUNNING) {
                            targetState = State.Type.SUCCESS;
                        } else if (newState == State.Type.KILLING) {
                            targetState = State.Type.KILLED;
                        }
                    } else {
                        // we should set the state to RUNNING so that subtasks are executed