Repository: kestra-io/kestra (java)

Issue:
Unable to trigger execution with large file in input parameters
### Describe the issue

I get the following error when I try to trigger an execution with a large file supplied as an input parameter.

```
java.lang.ArithmeticException: integer overflow
	at java.base/java.lang.Math.toIntExact(Unknown Source)
	at io.micronaut.http.server.netty.MicronautHttpData.get(MicronautHttpData.java:360)
	at io.micronaut.http.server.netty.MicronautHttpData.getString(MicronautHttpData.java:411)
	at io.micronaut.http.server.netty.body.ImmediateMultiObjectBody.toMap(ImmediateMultiObjectBody.java:103)
	at io.micronaut.http.server.netty.FormRouteCompleter.asMap(FormRouteCompleter.java:245)
	at io.micronaut.http.server.netty.NettyHttpRequest.getBody(NettyHttpRequest.java:402)
	at io.micronaut.http.server.netty.NettyHttpRequest.getBody(NettyHttpRequest.java:416)
	at io.micronaut.http.HttpMessage.getBody(HttpMessage.java:94)
	at io.micronaut.http.server.netty.NettyHttpRequest.getBody(NettyHttpRequest.java:410)
	at io.kestra.webserver.controllers.ExecutionController.create(ExecutionController.java:551)
	at io.kestra.ee.webserver.controllers.ExecutionController.create(ExecutionController.java:230)
	at io.kestra.ee.webserver.controllers.$ExecutionController$Definition$Intercepted.$$access$$create(Unknown Source)
	at io.kestra.ee.webserver.controllers.$ExecutionController$Definition$Exec.dispatch(Unknown Source)
	at io.micronaut.context.AbstractExecutableMethodsDefinition$DispatchedExecutableMethod.invoke(AbstractExecutableMethodsDefinition.java:456)
	at io.micronaut.aop.chain.MethodInterceptorChain.proceed(MethodInterceptorChain.java:128)
	at io.micronaut.validation.ValidatingInterceptor.validateReturnMicronautValidator(ValidatingInterceptor.java:153)
	at io.micronaut.validation.ValidatingInterceptor.intercept(ValidatingInterceptor.java:139)
	at io.micronaut.aop.chain.MethodInterceptorChain.proceed(MethodInterceptorChain.java:137)
	at io.kestra.ee.webserver.controllers.$ExecutionController$Definition$Intercepted.create(Unknown Source)
	at io.kestra.ee.webserver.controllers.$ExecutionController$Definition$Exec.dispatch(Unknown Source)
	at io.micronaut.context.AbstractExecutableMethodsDefinition$DispatchedExecutableMethod.invokeUnsafe(AbstractExecutableMethodsDefinition.java:461)
	at io.micronaut.context.DefaultBeanContext$BeanContextUnsafeExecutionHandle.invokeUnsafe(DefaultBeanContext.java:4274)
	at io.micronaut.web.router.AbstractRouteMatch.execute(AbstractRouteMatch.java:271)
	at io.micronaut.http.server.RouteExecutor.executeRouteAndConvertBody(RouteExecutor.java:488)
	at io.micronaut.http.server.RouteExecutor.lambda$callRoute$6(RouteExecutor.java:465)
	at io.micronaut.core.execution.ExecutionFlow.lambda$async$1(ExecutionFlow.java:87)
	at io.micrometer.core.instrument.composite.CompositeTimer.record(CompositeTimer.java:141)
	at io.micrometer.core.instrument.Timer.lambda$wrap$0(Timer.java:193)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
	at java.base/java.lang.Thread.run(Unknown Source)
```

I have the same error, whether execution is triggered by the UI or by a curl.

```
curl "https://kestra/api/v1/executions/***" -H "ContentType: multipart/form-data" -F "files=@file.ion;filename=from"
```

### Environment

- Kestra Version: EE 0.15.9
- Operating System: Kubernetes (ES/Kafka/GCS)


Pull request:
fix(core): file upload
Fixes #3419
Fixes https://github.com/kestra-io/kestra/issues/3386

Patch:
diff --git a/build.gradle b/build.gradle
index 8e09c7fda..4cb2f75d7 100644
--- a/build.gradle
+++ b/build.gradle
@@ -114,6 +114,9 @@ allprojects {
 
             // issue with the Docker lib having a too old version for the k8s extension
             force("org.apache.commons:commons-compress:1.26.1")
+
+            // the new version brings a bugfix for large file upload
+            force("io.micronaut:micronaut-http-server-netty:4.4.9")
         }
     }
 
diff --git a/cli/src/main/resources/application.yml b/cli/src/main/resources/application.yml
index 24dcc6452..f533ba5df 100644
--- a/cli/src/main/resources/application.yml
+++ b/cli/src/main/resources/application.yml
@@ -16,7 +16,7 @@ micronaut:
     max-request-size: 10GB
     multipart:
       max-file-size: 10GB
-      disk: true
+      mixed: true
     read-idle-timeout: 60m
     write-idle-timeout: 60m
     idle-timeout: 60m
diff --git a/core/build.gradle b/core/build.gradle
index 2017b1768..a853131dd 100644
--- a/core/build.gradle
+++ b/core/build.gradle
@@ -21,6 +21,9 @@ dependencies {
     // serializers
     implementation group: 'com.fasterxml.jackson.dataformat', name: 'jackson-dataformat-ion', version: jacksonVersion
 
+    // micronaut
+    implementation "io.micronaut:micronaut-http-server-netty"
+
     // utils
     implementation 'com.github.oshi:oshi-core:6.6.0'
     implementation 'io.pebbletemplates:pebble:3.2.2'
diff --git a/core/src/main/java/io/kestra/core/runners/FlowInputOutput.java b/core/src/main/java/io/kestra/core/runners/FlowInputOutput.java
index 1de04374d..aa8db6683 100644
--- a/core/src/main/java/io/kestra/core/runners/FlowInputOutput.java
+++ b/core/src/main/java/io/kestra/core/runners/FlowInputOutput.java
@@ -15,15 +15,16 @@ import io.kestra.core.serializers.JacksonMapper;
 import io.kestra.core.storages.StorageInterface;
 import io.micronaut.context.annotation.Value;
 import io.micronaut.core.annotation.Nullable;
-import io.micronaut.http.multipart.StreamingFileUpload;
+import io.micronaut.http.multipart.CompletedFileUpload;
+import io.micronaut.http.multipart.CompletedPart;
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Schedulers;
 
 import java.io.File;
+import java.io.FileOutputStream;
 import java.io.IOException;
 import java.net.URI;
 import java.security.GeneralSecurityException;
@@ -61,82 +62,69 @@ public class FlowInputOutput {
     /**
      * Utility method for retrieving types inputs for a flow.
      *
-     * @param flow      The Flow
+     * @param flow      The Flow.
      * @param execution The Execution.
-     * @param in        The Flow's inputs.
+     * @param inputs        The Flow's inputs.
      * @return The Map of typed inputs.
      */
-    public Map<String, Object> typedInputs(
-        final Flow flow,
-        final Execution execution,
-        final Map<String, Object> in,
-        final Publisher<StreamingFileUpload> files
-    ) throws IOException {
-        return this.typedInputs(
-            flow.getInputs(),
-            execution,
-            in,
-            files
-        );
+    public Map<String, Object> typedInputs(final Flow flow,
+                                           final Execution execution,
+                                           final Publisher<CompletedPart> inputs) throws IOException {
+        return this.typedInputs(flow.getInputs(), execution, inputs);
     }
 
     /**
-     * Utility method for retrieving types inputs.
+     * Utility method for retrieving types inputs for a flow.
      *
-     * @param inputs    The Inputs.
+     * @param inputs      The inputs
      * @param execution The Execution.
-     * @param in        The Flow's inputs.
+     * @param in        The Execution's inputs.
      * @return The Map of typed inputs.
      */
-    public Map<String, Object> typedInputs(
-        final List<Input<?>> inputs,
-        final Execution execution,
-        final Map<String, Object> in,
-        final Publisher<StreamingFileUpload> files
-    ) throws IOException {
-        if (files == null) {
-            return this.typedInputs(inputs, execution, in);
-        }
-
-        Map<String, String> uploads = Flux.from(files)
+    public Map<String, Object> typedInputs(final List<Input<?>> inputs,
+                                           final Execution execution,
+                                           final Publisher<CompletedPart> in) throws IOException {
+        Map<String, Object> uploads = Flux.from(in)
             .subscribeOn(Schedulers.boundedElastic())
-            .map(throwFunction(file -> {
-                File tempFile = File.createTempFile(file.getFilename() + "_", ".upl");
-                Publisher<Boolean> uploadPublisher = file.transferTo(tempFile);
-                Boolean bool = Mono.from(uploadPublisher).block();
-
-                if (Boolean.FALSE.equals(bool)) {
-                    throw new RuntimeException("Can't upload");
-                }
+            .map(throwFunction(input -> {
+                if (input instanceof CompletedFileUpload fileUpload) {
+                    File tempFile = File.createTempFile(fileUpload.getFilename() + "_", ".upl");
+                    try (var inputStream = fileUpload.getInputStream();
+                         var outputStream = new FileOutputStream(tempFile)) {
+                        long transferredBytes = inputStream.transferTo(outputStream);
+                        if (transferredBytes == 0) {
+                            throw new RuntimeException("Can't upload file: " + fileUpload.getFilename());
+                        }
+                    }
+                    URI from = storageInterface.from(execution, fileUpload.getFilename(), tempFile);
+                    if (!tempFile.delete()) {
+                        tempFile.deleteOnExit();
+                    }
 
-                URI from = storageInterface.from(execution, file.getFilename(), tempFile);
-                //noinspection ResultOfMethodCallIgnored
-                tempFile.delete();
+                    return new AbstractMap.SimpleEntry<>(
+                        fileUpload.getFilename(),
+                        (Object) from.toString()
+                    );
 
-                return new AbstractMap.SimpleEntry<>(
-                    file.getFilename(),
-                    from.toString()
-                );
+                } else {
+                    return new AbstractMap.SimpleEntry<>(
+                        input.getName(),
+                        (Object) new String(input.getBytes())
+                    );
+                }
             }))
             .collectMap(AbstractMap.SimpleEntry::getKey, AbstractMap.SimpleEntry::getValue)
             .block();
 
-        Map<String, Object> merged = new HashMap<>();
-        if (in != null) {
-            merged.putAll(in);
-        }
-
-        merged.putAll(uploads);
-
-        return this.typedInputs(inputs, execution, merged);
+        return this.typedInputs(inputs, execution, uploads);
     }
 
     /**
      * Utility method for retrieving types inputs for a flow.
      *
-     * @param flow      The inputs Flow?
+     * @param flow      The Flow.
      * @param execution The Execution.
-     * @param in        The Flow's inputs.
+     * @param in        The Execution's inputs.
      * @return The Map of typed inputs.
      */
     public Map<String, Object> typedInputs(
@@ -156,7 +144,7 @@ public class FlowInputOutput {
      *
      * @param inputs    The inputs.
      * @param execution The Execution.
-     * @param in        The Flow's inputs.
+     * @param in        The Execution's inputs.
      * @return The Map of typed inputs.
      */
     @SuppressWarnings({"unchecked", "rawtypes"})
diff --git a/core/src/main/java/io/kestra/core/runners/RunnerUtils.java b/core/src/main/java/io/kestra/core/runners/RunnerUtils.java
index 6fd181854..28ae59600 100644
--- a/core/src/main/java/io/kestra/core/runners/RunnerUtils.java
+++ b/core/src/main/java/io/kestra/core/runners/RunnerUtils.java
@@ -9,13 +9,22 @@ import io.kestra.core.queues.QueueInterface;
 import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.services.ConditionService;
 import io.kestra.core.utils.Await;
-import io.micronaut.http.multipart.StreamingFileUpload;
+import io.kestra.core.utils.IdUtils;
+import io.micronaut.context.annotation.Value;
+import io.micronaut.http.multipart.CompletedFileUpload;
+import io.micronaut.http.multipart.CompletedPart;
+import io.micronaut.http.server.netty.multipart.NettyCompletedAttribute;
+import io.micronaut.http.server.netty.multipart.NettyCompletedFileUpload;
 import jakarta.inject.Inject;
 import jakarta.inject.Named;
 import jakarta.inject.Singleton;
 import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Schedulers;
 
-import java.io.IOException;
+import java.io.*;
+import java.net.URI;
+import java.security.GeneralSecurityException;
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
@@ -43,8 +52,8 @@ public class RunnerUtils {
     private FlowInputOutput flowIO;
 
     @Deprecated
-    public Map<String, Object> typedInputs(Flow flow, Execution execution, Map<String, Object> in, Publisher<StreamingFileUpload> files) throws IOException {
-        return flowIO.typedInputs(flow, execution, in, files);
+    public Map<String, Object> typedInputs(Flow flow, Execution execution, Publisher<CompletedPart> inputs) throws IOException {
+        return flowIO.typedInputs(flow, execution, inputs);
     }
 
     @Deprecated
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 cefc72ad2..8969ece9f 100644
--- a/core/src/main/java/io/kestra/core/services/ExecutionService.java
+++ b/core/src/main/java/io/kestra/core/services/ExecutionService.java
@@ -25,6 +25,7 @@ import io.kestra.core.utils.GraphUtils;
 import io.kestra.core.utils.IdUtils;
 import io.micronaut.context.event.ApplicationEventPublisher;
 import io.micronaut.core.annotation.Nullable;
+import io.micronaut.http.multipart.CompletedPart;
 import io.micronaut.http.multipart.StreamingFileUpload;
 import jakarta.inject.Inject;
 import jakarta.inject.Named;
@@ -264,10 +265,10 @@ public class ExecutionService {
     }
 
     public Execution markAs(final Execution execution, Flow flow, String taskRunId, State.Type newState) throws Exception {
-        return this.markAs(execution, flow, taskRunId, newState, null, null);
+        return this.markAs(execution, flow, taskRunId, newState, null);
     }
 
-    private Execution markAs(final Execution execution, Flow flow, String taskRunId, State.Type newState, @Nullable Map<String, Object> onResumeInputs, @Nullable Publisher<StreamingFileUpload> onResumeFiles) throws Exception {
+    private Execution markAs(final Execution execution, Flow flow, String taskRunId, State.Type newState, @Nullable Map<String, Object> onResumeInputs) throws Exception {
         Set<String> taskRunToRestart = this.taskRunToRestart(
             execution,
             taskRun -> taskRun.getId().equals(taskRunId)
@@ -284,14 +285,7 @@ public class ExecutionService {
                 TaskRun newTaskRun = originalTaskRun.withState(newState);
 
                 if (task instanceof Pause pauseTask && pauseTask.getOnResume() != null) {
-                    Map<String, Object> pauseOutputs = flowInputOutput.typedInputs(
-                        pauseTask.getOnResume(),
-                        execution,
-                        onResumeInputs,
-                        onResumeFiles
-                    );
-
-                    newTaskRun = newTaskRun.withOutputs(pauseTask.generateOutputs(pauseOutputs));
+                    newTaskRun = newTaskRun.withOutputs(pauseTask.generateOutputs(onResumeInputs));
                 }
 
                 if (task instanceof Pause pauseTask && pauseTask.getTasks() == null && newState == State.Type.RUNNING) {
@@ -402,7 +396,7 @@ public class ExecutionService {
      * @throws Exception if the state of the execution cannot be updated
      */
     public Execution resume(Execution execution, Flow flow, State.Type newState) throws Exception {
-        return this.resume(execution, flow, newState, null, null);
+        return this.resume(execution, flow, newState, (Map<String, Object>) null);
     }
 
     /**
@@ -410,19 +404,47 @@ public class ExecutionService {
      * The execution must be paused or this call will be a no-op.
      *
      * @param execution the execution to resume
-     * @param newState  should be RUNNING or KILLING, other states may lead to undefined behaviour
+     * @param newState  should be RUNNING or KILLING, other states may lead to undefined behavior
+     * @param flow      the flow of the execution
+     * @param inputs    the onResume inputs
+     * @return the execution in the new state.
+     * @throws Exception if the state of the execution cannot be updated
+     */
+    public Execution resume(final Execution execution, Flow flow, State.Type newState, @Nullable Publisher<CompletedPart> inputs) throws Exception {
+        var runningTaskRun = execution
+            .findFirstByState(State.Type.PAUSED)
+            .orElseThrow(() -> new IllegalArgumentException("No paused task found on execution " + execution.getId()));
+
+        var task = flow.findTaskByTaskId(runningTaskRun.getTaskId());
+        Map<String, Object> pauseOutputs = Collections.emptyMap();
+        if (task instanceof Pause pauseTask) {
+            pauseOutputs = flowInputOutput.typedInputs(
+                pauseTask.getOnResume(),
+                execution,
+                inputs
+            );
+        }
+
+        return resume(execution, flow, newState, pauseOutputs);
+    }
+
+    /**
+     * Resume a paused execution to a new state.
+     * The execution must be paused or this call will be a no-op.
+     *
+     * @param execution the execution to resume
+     * @param newState  should be RUNNING or KILLING, other states may lead to undefined behavior
      * @param flow      the flow of the execution
      * @param inputs    the onResume inputs
-     * @param files     the onResume files
      * @return the execution in the new state.
      * @throws Exception if the state of the execution cannot be updated
      */
-    public Execution resume(final Execution execution, Flow flow, State.Type newState, @Nullable Map<String, Object> inputs, @Nullable Publisher<StreamingFileUpload> files) throws Exception {
+    public Execution resume(final Execution execution, Flow flow, State.Type newState, @Nullable Map<String, Object> inputs) throws Exception {
         var runningTaskRun = execution
             .findFirstByState(State.Type.PAUSED)
             .orElseThrow(() -> new IllegalArgumentException("No paused task found on execution " + execution.getId()));
 
-        var unpausedExecution = this.markAs(execution, flow, runningTaskRun.getId(), newState, inputs, files);
+        var unpausedExecution = this.markAs(execution, flow, runningTaskRun.getId(), newState, inputs);
 
         this.eventPublisher.publishEvent(new CrudEvent<>(execution, CrudEventType.UPDATE));
         return unpausedExecution;
diff --git a/core/src/main/java/io/kestra/plugin/core/execution/Resume.java b/core/src/main/java/io/kestra/plugin/core/execution/Resume.java
index 11f22e658..3bdc83e87 100644
--- a/core/src/main/java/io/kestra/plugin/core/execution/Resume.java
+++ b/core/src/main/java/io/kestra/plugin/core/execution/Resume.java
@@ -32,7 +32,7 @@ import java.util.Map;
 @Getter
 @NoArgsConstructor
 @Schema(
-    title = "Resume a paused execution."
+    title = "Resume a paused execution. By default, the task assumes that you want to resume the current `executionId`. If you want to programmatically resume an execution of another flow, make sure to define the `executionId`, `flowId`, and `namespace` properties explicitly. Using the `inputs` property, you can additionally pass custom `onResume` input values to the execution."
 )
 @Plugin(
     examples = {
@@ -45,7 +45,7 @@ import java.util.Map;
 )
 public class Resume  extends Task implements RunnableTask<VoidOutput> {
     @Schema(
-        title = "Filter for a specific namespace in case `executionId` is set."
+        title = "Filter for a specific namespace in case `executionId` is set. In case you wonder why `executionId` is not enough — we require specifying the namespace to make permissions explicit. The Enterprise Edition of Kestra allows you to resume executions from another namespaces only if the permissions allow it. Check the [Allowed Namespaces](https://kestra.io/docs/enterprise/allowed-namespaces) documentation for more details."
     )
     @PluginProperty(dynamic = true)
     private String namespace;
@@ -59,8 +59,11 @@ public class Resume  extends Task implements RunnableTask<VoidOutput> {
     @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."""
+            If you explicitly define an `executionId`, Kestra will use that specific ID.
+            
+            If another `namespace` and `flowId` properties are set, Kestra will look for a paused execution for that corresponding flow.
+            
+            If `executionId` is not set, the task will use the ID of the current execution."""
     )
     @PluginProperty(dynamic = true)
     private String executionId;
@@ -85,7 +88,7 @@ public class Resume  extends Task implements RunnableTask<VoidOutput> {
             .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);
+        Execution resumed = executionService.resume(execution, flow, State.Type.RUNNING, renderedInputs);
         executionQueue.emit(resumed);
 
         return null;
diff --git a/gradle.properties b/gradle.properties
index da032ce05..f2bb2dee9 100644
--- a/gradle.properties
+++ b/gradle.properties
@@ -1,7 +1,8 @@
 version=0.17.0-SNAPSHOT
 
 jacksonVersion=2.16.2
-micronautVersion=4.4.1
+# warning, next time Micronaut is updated, the force of micronaut-core in build.gradle must be removed
+micronautVersion=4.4.2
 lombokVersion=1.18.32
 slf4jVersion=2.0.13
 
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 081f30be4..e7b982485 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
@@ -62,6 +62,7 @@ import io.micronaut.http.annotation.Put;
 import io.micronaut.http.annotation.QueryValue;
 import io.micronaut.http.exceptions.HttpStatusException;
 import io.micronaut.http.multipart.StreamingFileUpload;
+import io.micronaut.http.server.multipart.MultipartBody;
 import io.micronaut.http.server.types.files.StreamedFile;
 import io.micronaut.http.sse.Event;
 import io.micronaut.scheduling.TaskExecutors;
@@ -518,14 +519,13 @@ public class ExecutionController {
     @Deprecated
     public Execution trigger(
         @Parameter(description = "The flow namespace") @PathVariable String namespace,
-        @Parameter(description = "The flow id") @PathVariable String id,
-        @Parameter(description = "The inputs") HttpRequest<?> inputs,
+        @Parameter(description = "The flow id") @Nullable @PathVariable String id,
+        @Parameter(description = "The inputs") @Nullable  @Body MultipartBody inputs,
         @Parameter(description = "The labels as a list of 'key:value'") @Nullable @QueryValue List<String> labels,
-        @Parameter(description = "The inputs of type file") @Nullable @Part Publisher<StreamingFileUpload> files,
         @Parameter(description = "If the server will wait the end of the execution") @QueryValue(defaultValue = "false") Boolean wait,
         @Parameter(description = "The flow revision or latest if null") @QueryValue Optional<Integer> revision
     ) throws IOException {
-        return this.create(namespace, id, inputs, labels, files, wait, revision);
+        return this.create(namespace, id, inputs, labels, wait, revision);
     }
 
     @ExecuteOn(TaskExecutors.IO)
@@ -535,9 +535,8 @@ public class ExecutionController {
     public Execution create(
         @Parameter(description = "The flow namespace") @PathVariable String namespace,
         @Parameter(description = "The flow id") @PathVariable String id,
-        @Parameter(description = "The inputs") HttpRequest<?> inputs, // FIXME we had to inject the HttpRequest here due to https://github.com/micronaut-projects/micronaut-core/issues/9694
+        @Parameter(description = "The inputs") @Nullable @Body MultipartBody inputs,
         @Parameter(description = "The labels as a list of 'key:value'") @Nullable @QueryValue List<String> labels,
-        @Parameter(description = "The inputs of type file") @Nullable @Part Publisher<StreamingFileUpload> files,
         @Parameter(description = "If the server will wait the end of the execution") @QueryValue(defaultValue = "false") Boolean wait,
         @Parameter(description = "The flow revision or latest if null") @QueryValue Optional<Integer> revision
     ) throws IOException {
@@ -555,10 +554,9 @@ public class ExecutionController {
             throw new IllegalStateException("Cannot execute an invalid flow: " + fwe.getException());
         }
 
-        Map<String, Object> inputMap = (Map<String, Object>) inputs.getBody(Map.class).orElse(null);
         Execution current = Execution.newExecution(
             found,
-            throwBiFunction((flow, execution) -> flowInputOutput.typedInputs(flow, execution, inputMap, files)),
+            throwBiFunction((flow, execution) -> flowInputOutput.typedInputs(flow, execution, inputs)),
             parseLabels(labels)
         );
 
@@ -959,8 +957,7 @@ public class ExecutionController {
     @ApiResponse(responseCode = "409", description = "if the executions is not paused")
     public HttpResponse<?> resume(
         @Parameter(description = "The execution id") @PathVariable String executionId,
-        @Parameter(description = "The inputs") HttpRequest<?> inputs, // FIXME we had to inject the HttpRequest here due to https://github.com/micronaut-projects/micronaut-core/issues/9694
-        @Parameter(description = "The inputs of type file") @Nullable @Part Publisher<StreamingFileUpload> files
+        @Parameter(description = "The inputs") @Nullable @Body MultipartBody inputs
     ) throws Exception {
         Optional<Execution> maybeExecution = executionRepository.findById(tenantService.resolveTenant(), executionId);
         if (maybeExecution.isEmpty()) {
@@ -974,9 +971,7 @@ public class ExecutionController {
 
         var flow = flowRepository.findByExecutionWithoutAcl(execution);
 
-        Map<String, Object> inputMap = (Map<String, Object>) inputs.getBody(Map.class).orElse(null);
-
-        Execution resumeExecution = this.executionService.resume(execution, flow, State.Type.RUNNING, inputMap, files);
+        Execution resumeExecution = this.executionService.resume(execution, flow, State.Type.RUNNING, inputs);
         this.executionQueue.emit(resumeExecution);
         return HttpResponse.noContent();
     }


Test patch:
diff --git a/core/src/test/java/io/kestra/core/runners/TestStreamingFileUpload.java b/core/src/test/java/io/kestra/core/runners/TestStreamingFileUpload.java
deleted file mode 100644
index f914de40f..000000000
--- a/core/src/test/java/io/kestra/core/runners/TestStreamingFileUpload.java
+++ /dev/null
@@ -1,148 +0,0 @@
-package io.kestra.core.runners;
-
-import io.micronaut.core.annotation.Nullable;
-import io.micronaut.http.MediaType;
-import io.micronaut.http.multipart.*;
-import org.apache.commons.lang3.NotImplementedException;
-import org.reactivestreams.Publisher;
-import org.reactivestreams.Subscriber;
-import reactor.core.publisher.Flux;
-
-import java.io.ByteArrayInputStream;
-import java.io.File;
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.ByteBuffer;
-import java.nio.file.Files;
-import java.util.Optional;
-
-public class TestStreamingFileUpload implements StreamingFileUpload {
-    private final String name;
-    private final MediaType contentType;
-    private final byte[] data;
-
-    public TestStreamingFileUpload(String name, byte[] data, @Nullable MediaType contentType) {
-        this.name = name;
-        this.contentType = contentType;
-        this.data = data;
-    }
-
-    @Override
-    public Publisher<Boolean> transferTo(String location) {
-        throw new NotImplementedException("transferTo(String location) is not implemented");
-    }
-
-    @SuppressWarnings("BlockingMethodInNonBlockingContext")
-    @Override
-    public Publisher<Boolean> transferTo(File destination) {
-        return Flux.create(fluxSink -> {
-            try {
-                Files.write(destination.toPath(), this.data);
-                fluxSink.next(true);
-            } catch (IOException e) {
-                fluxSink.error(e);
-            }
-        });
-    }
-
-    @Override
-    public Publisher<Boolean> delete() {
-        throw new NotImplementedException("delete() is not implemented");
-    }
-
-    @Override
-    public Optional<MediaType> getContentType() {
-        return Optional.ofNullable(this.contentType);
-    }
-
-    @Override
-    public String getName() {
-        return this.name;
-    }
-
-    @Override
-    public String getFilename() {
-        return this.name;
-    }
-
-    @Override
-    public long getSize() {
-        return this.data.length;
-    }
-
-    @Override
-    public long getDefinedSize() {
-        return this.data.length;
-    }
-
-    @Override
-    public boolean isComplete() {
-        return true;
-    }
-
-    @Override
-    public void subscribe(Subscriber<? super PartData> s) {
-        s.onNext(new CompletedFileUpload(
-            this.name,
-            this.data,
-            this.contentType
-        ));
-    }
-
-    public static class CompletedFileUpload implements io.micronaut.http.multipart.CompletedFileUpload {
-        private final String name;
-        private final MediaType contentType;
-        private final byte[] data;
-
-        public CompletedFileUpload(String name, byte[] data, @Nullable MediaType contentType) {
-            this.name = name;
-            this.contentType = contentType;
-            this.data = data;
-        }
-
-        @Override
-        public InputStream getInputStream() throws IOException {
-            return new ByteArrayInputStream(this.data);
-        }
-
-        @Override
-        public byte[] getBytes() throws IOException {
-            return this.data;
-        }
-
-        @Override
-        public ByteBuffer getByteBuffer() throws IOException {
-            return ByteBuffer.wrap(this.data);
-        }
-
-        @Override
-        public Optional<MediaType> getContentType() {
-            return Optional.ofNullable(this.contentType);
-        }
-
-        @Override
-        public String getName() {
-            return this.name;
-        }
-
-        @Override
-        public String getFilename() {
-            return this.name;
-        }
-
-        @Override
-        public long getSize() {
-            return this.data.length;
-        }
-
-        @Override
-        public long getDefinedSize() {
-            return this.data.length;
-        }
-
-        @Override
-        public boolean isComplete() {
-            return true;
-        }
-    }
-}
diff --git a/core/src/test/java/io/kestra/plugin/core/flow/PauseTest.java b/core/src/test/java/io/kestra/plugin/core/flow/PauseTest.java
index 574578df8..35b3741b8 100644
--- a/core/src/test/java/io/kestra/plugin/core/flow/PauseTest.java
+++ b/core/src/test/java/io/kestra/plugin/core/flow/PauseTest.java
@@ -10,19 +10,27 @@ import io.kestra.core.queues.QueueInterface;
 import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.runners.AbstractMemoryRunnerTest;
 import io.kestra.core.runners.RunnerUtils;
-import io.kestra.core.runners.TestStreamingFileUpload;
 import io.kestra.core.services.ExecutionService;
 import io.kestra.core.storages.StorageInterface;
 import io.micronaut.http.MediaType;
+import io.micronaut.http.multipart.CompletedPart;
+import io.micronaut.http.server.HttpServerConfiguration;
+import io.micronaut.http.server.netty.MicronautHttpData;
+import io.micronaut.http.server.netty.multipart.NettyCompletedAttribute;
+import io.micronaut.http.server.netty.multipart.NettyCompletedFileUpload;
+import io.netty.buffer.Unpooled;
+import io.netty.handler.codec.http.multipart.*;
 import jakarta.inject.Inject;
 import jakarta.inject.Named;
 import jakarta.inject.Singleton;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 import java.io.InputStreamReader;
 import java.net.URI;
+import java.nio.charset.Charset;
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
@@ -195,12 +203,17 @@ public class PauseTest extends AbstractMemoryRunnerTest {
             assertThat(execution.getTaskRunList().get(0).getState().getCurrent(), is(State.Type.PAUSED));
             assertThat(execution.getTaskRunList(), hasSize(1));
 
+            CompletedPart part1 = new NettyCompletedAttribute(new MemoryAttribute("asked", "restarted"));
+            byte[] data = executionId.getBytes();
+            HttpDataFactory httpDataFactory = new MicronautHttpData.Factory(new HttpServerConfiguration.MultipartConfiguration(), null);
+            FileUpload fileUpload = httpDataFactory.createFileUpload(null, "files", "data", MediaType.TEXT_PLAIN, null, Charset.defaultCharset(), data.length);
+            fileUpload.addContent(Unpooled.copiedBuffer(data), true);
+            CompletedPart part2 = new NettyCompletedFileUpload(fileUpload);
             Execution restarted = executionService.resume(
                 execution,
                 flow,
                 State.Type.RUNNING,
-                Map.of("asked", "restarted"),
-                Flux.just(new TestStreamingFileUpload("data", executionId.getBytes(), MediaType.TEXT_PLAIN_TYPE))
+                Flux.just(part1, part2)
             );
 
             execution = runnerUtils.awaitExecution(
@@ -228,7 +241,7 @@ public class PauseTest extends AbstractMemoryRunnerTest {
 
             MissingRequiredArgument e = assertThrows(
                 MissingRequiredArgument.class,
-                () -> executionService.resume(execution, flow, State.Type.RUNNING)
+                () -> executionService.resume(execution, flow, State.Type.RUNNING, Mono.empty())
             );
 
             assertThat(e.getMessage(), containsString("required input value 'asked'"));