Repository: kestra-io/kestra (java)

Issue:
Add a `ResumeExecution` execution task to resume paused execution by ID
After pausing an execution with [Pause](https://kestra.io/plugins/core/tasks/flows/io.kestra.core.tasks.flows.Pause.html) task If you want to unpaused it, for now you can only handle it an api call or from the ui. 

We should add `Unpaused` task to do it directly from Kestra flow. By this way, we will be able to use for an example a trigger from a queue and to restart the execution based on the current message. 
The design on the task can be tricky since : 
- if the trigger is from the queue, we need to support internal storage 
- if the trigger is from another task, we need to support raw data from outputs  

Related discussion in [slack](https://kestra-io.slack.com/archives/C03FQKXRK3K/p1677689218600379)


Pull request description:
feat: task to resume an execution
Fixes #1041

You can test it on a single flow but I wouldn't recommend that on production ;) 

```yaml
id: resumeMyself
namespace: myteam

tasks:
  - id: parallel
    type: io.kestra.core.tasks.flows.Parallel
    tasks:
    - id: pause
      type: io.kestra.core.tasks.flows.Pause
    - id: sequential
      type: io.kestra.core.tasks.flows.Sequential
      tasks:
        - id: delay
          type: io.kestra.core.tasks.flows.Pause
          delay: PT1S
        - id: resume
          type: io.kestra.core.tasks.executions.Resume
```

An example with an input that must correspond to the `onResume` input of the target flow Pause task:
```yaml
id: resumeExecution
namespace: myteam

tasks:
  - id: resume
    type: io.kestra.core.tasks.executions.Resume
    executionId: 46T32z9xVLZH8vpwd43qXo
    flowId: pause
    namespace: myteam
    inputs:
      input: input
```

Patch:
diff --git a/core/src/main/java/io/kestra/core/tasks/executions/Resume.java b/core/src/main/java/io/kestra/core/tasks/executions/Resume.java
new file mode 100644
index 000000000..22914e0d5
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/tasks/executions/Resume.java
@@ -0,0 +1,93 @@
+package io.kestra.core.tasks.executions;
+
+import io.kestra.core.models.annotations.Example;
+import io.kestra.core.models.annotations.Plugin;
+import io.kestra.core.models.annotations.PluginProperty;
+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.tasks.RunnableTask;
+import io.kestra.core.models.tasks.Task;
+import io.kestra.core.models.tasks.VoidOutput;
+import io.kestra.core.queues.QueueFactoryInterface;
+import io.kestra.core.queues.QueueInterface;
+import io.kestra.core.repositories.ExecutionRepositoryInterface;
+import io.kestra.core.runners.FlowExecutorInterface;
+import io.kestra.core.runners.RunContext;
+import io.kestra.core.services.ExecutionService;
+import io.kestra.core.tasks.PluginUtilsService;
+import io.micronaut.inject.qualifiers.Qualifiers;
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+import lombok.experimental.SuperBuilder;
+
+import java.util.Map;
+
+@SuperBuilder
+@ToString
+@EqualsAndHashCode
+@Getter
+@NoArgsConstructor
+@Schema(
+    title = "Resume a paused execution."
+)
+@Plugin(
+    examples = {
+        @Example(
+            code = {
+                "executionId: \"{{ trigger.executionId }}\""
+            }
+        )
+    }
+)
+public class Resume  extends Task implements RunnableTask<VoidOutput> {
+    @Schema(
+        title = "Filter for a specific namespace in case `executionId` is set."
+    )
+    @PluginProperty(dynamic = true)
+    private String namespace;
+
+    @Schema(
+        title = "Filter for a specific flow identifier in case `executionId` is set."
+    )
+    @PluginProperty(dynamic = true)
+    private String flowId;
+
+    @Schema(
+        title = "Filter for a specific execution.",
+        description = """
+            If not set, the task will use the ID of the current execution.
+            If set, it will try to locate the execution on the current flow unless the `namespace` and `flowId` properties are set."""
+    )
+    @PluginProperty(dynamic = true)
+    private String executionId;
+
+    @Schema(
+        title = "Inputs to be passed to the execution when it's resumed."
+    )
+    @PluginProperty(dynamic = true)
+    private Map<String, Object> inputs;
+
+    @Override
+    public VoidOutput run(RunContext runContext) throws Exception {
+        System.out.println("resume");
+        var executionInfo = PluginUtilsService.executionFromTaskParameters(runContext, this.namespace, this.flowId, this.executionId);
+
+        ExecutionService executionService = runContext.getApplicationContext().getBean(ExecutionService.class);
+        ExecutionRepositoryInterface executionRepository = runContext.getApplicationContext().getBean(ExecutionRepositoryInterface.class);
+        FlowExecutorInterface flowExecutor = runContext.getApplicationContext().getBean(FlowExecutorInterface.class);
+        QueueInterface<Execution> executionQueue = runContext.getApplicationContext().getBean(QueueInterface.class, Qualifiers.byName(QueueFactoryInterface.EXECUTION_NAMED));
+
+        Execution execution = executionRepository.findById(executionInfo.tenantId(), executionInfo.id())
+            .orElseThrow(() -> new IllegalArgumentException("No execution found for execution id " + executionInfo.id()));
+        Flow flow = flowExecutor.findByExecution(execution).orElseThrow(() -> new IllegalArgumentException("Flow not found for execution id " + executionInfo.id()));
+        Map<String, Object> renderedInputs = inputs != null ? runContext.render(inputs) : null;
+        Execution resumed = executionService.resume(execution, flow, State.Type.RUNNING, renderedInputs, null);
+        executionQueue.emit(resumed);
+
+        return null;
+    }
+}


Test patch:
diff --git a/core/src/test/java/io/kestra/core/tasks/executions/ResumeTest.java b/core/src/test/java/io/kestra/core/tasks/executions/ResumeTest.java
new file mode 100644
index 000000000..03555438e
--- /dev/null
+++ b/core/src/test/java/io/kestra/core/tasks/executions/ResumeTest.java
@@ -0,0 +1,20 @@
+package io.kestra.core.tasks.executions;
+
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.State;
+import io.kestra.core.runners.AbstractMemoryRunnerTest;
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.is;
+
+class ResumeTest extends AbstractMemoryRunnerTest {
+    @Test
+    void resume() throws Exception {
+        Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "resume-execution");
+
+        assertThat(execution.getTaskRunList(), hasSize(5));
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
+    }
+}
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/resume-execution.yaml b/core/src/test/resources/flows/valids/resume-execution.yaml
new file mode 100644
index 000000000..a0fc7f2f2
--- /dev/null
+++ b/core/src/test/resources/flows/valids/resume-execution.yaml
@@ -0,0 +1,17 @@
+id: resume-execution
+namespace:  io.kestra.tests
+
+tasks:
+  - id: parallel
+    type: io.kestra.core.tasks.flows.Parallel
+    tasks:
+      - id: pause
+        type: io.kestra.core.tasks.flows.Pause
+      - id: sequential
+        type: io.kestra.core.tasks.flows.Sequential
+        tasks:
+          - id: log
+            type: io.kestra.core.tasks.test.Sleep
+            duration: 100
+          - id: resume
+            type: io.kestra.core.tasks.executions.Resume
\ No newline at end of file