Repository: kestra-io/kestra (java)

Issue:
Make "from" a required property in Split task
### Describe the issue

type: "io.kestra.core.tasks.storages.Split"

without from, the task cannot run so it must be mandatory

### Environment

- Kestra Version:
- Operating System (OS/Docker/Kubernetes):
- Java Version (if you don't run kestra in Docker):


Pull request:
feat(core): make the 'from' property of storage tasks mandatory
fixes #3274

Patch:
diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml
index 703ec1488..668ca6b1a 100644
--- a/.github/workflows/main.yml
+++ b/.github/workflows/main.yml
@@ -410,7 +410,7 @@ jobs:
 
       # Update
       - name: Update internal
-        uses: benc-uk/workflow-dispatch@v1.2.3
+        uses: benc-uk/workflow-dispatch@v1
         if: github.ref == 'refs/heads/develop'
         with:
           workflow: oss-build.yml
diff --git a/cli/src/main/java/io/kestra/cli/commands/flows/FlowExpandCommand.java b/cli/src/main/java/io/kestra/cli/commands/flows/FlowExpandCommand.java
index bb6e0ac27..aa885b69c 100644
--- a/cli/src/main/java/io/kestra/cli/commands/flows/FlowExpandCommand.java
+++ b/cli/src/main/java/io/kestra/cli/commands/flows/FlowExpandCommand.java
@@ -12,8 +12,9 @@ import java.nio.file.Path;
 
 @CommandLine.Command(
     name = "expand",
-    description = "expand a flow"
+    description = "deprecated - expand a flow"
 )
+@Deprecated
 public class FlowExpandCommand extends AbstractCommand {
 
     @CommandLine.Parameters(index = "0", description = "the flow file to expand")
@@ -28,6 +29,7 @@ public class FlowExpandCommand extends AbstractCommand {
     @Override
     public Integer call() throws Exception {
         super.call();
+        stdErr("Warning, this functionality is deprecated and will be removed at some point.");
         String content = IncludeHelperExpander.expand(Files.readString(file), file.getParent());
         Flow flow = yamlFlowParser.parse(content, Flow.class);
         modelValidator.validate(flow);
diff --git a/cli/src/main/java/io/kestra/cli/commands/flows/IncludeHelperExpander.java b/cli/src/main/java/io/kestra/cli/commands/flows/IncludeHelperExpander.java
index 2959ca0ca..7b130ccee 100644
--- a/cli/src/main/java/io/kestra/cli/commands/flows/IncludeHelperExpander.java
+++ b/cli/src/main/java/io/kestra/cli/commands/flows/IncludeHelperExpander.java
@@ -9,6 +9,7 @@ import java.nio.file.Path;
 import java.util.List;
 import java.util.stream.Collectors;
 
+@Deprecated
 public abstract class IncludeHelperExpander {
 
     public static String expand(String value, Path directory) throws IOException {
diff --git a/core/src/main/java/io/kestra/core/server/AbstractServiceLivenessCoordinator.java b/core/src/main/java/io/kestra/core/server/AbstractServiceLivenessCoordinator.java
index 501d19590..5e84e3171 100644
--- a/core/src/main/java/io/kestra/core/server/AbstractServiceLivenessCoordinator.java
+++ b/core/src/main/java/io/kestra/core/server/AbstractServiceLivenessCoordinator.java
@@ -51,7 +51,7 @@ public abstract class AbstractServiceLivenessCoordinator extends AbstractService
         // Multiple Executors can be running in parallel. We add a jitter to
         // help distributing the load more evenly among the ServiceLivenessCoordinator.
         // This is also used to prevent all ServiceLivenessCoordinator from attempting to query the repository simultaneously.
-        RandomGenerator r = RandomGenerator.getDefault();
+        Random r = new Random(); //SONAR
         int jitter = r.nextInt(DEFAULT_SCHEDULE_JITTER_MAX_MS);
         return serverConfig.liveness().interval().plus(Duration.ofMillis(jitter));
     }
diff --git a/core/src/main/java/io/kestra/core/tasks/flows/ForEachItem.java b/core/src/main/java/io/kestra/core/tasks/flows/ForEachItem.java
index d1e24e7fe..c0b868873 100644
--- a/core/src/main/java/io/kestra/core/tasks/flows/ForEachItem.java
+++ b/core/src/main/java/io/kestra/core/tasks/flows/ForEachItem.java
@@ -201,7 +201,7 @@ import static io.kestra.core.utils.Rethrow.throwFunction;
                     tasks:
                       - id: subflow_per_batch
                         type: io.kestra.core.tasks.flows.ForEachItem
-                        items: "{{ trigger.uris[parent.taskrun.value] }}" 
+                        items: "{{ trigger.uris[parent.taskrun.value] }}"
                         batch:
                           rows: 1
                         flowId: process_batch
@@ -354,6 +354,8 @@ public class ForEachItem extends Task implements FlowableTask<VoidOutput>, Child
     @Getter
     @NoArgsConstructor
     public static class ForEachItemSplit extends Task implements RunnableTask<ForEachItemSplit.Output> {
+        static final String SUFFIX = "_split";
+
         private String items;
         private Batch batch;
 
@@ -361,7 +363,7 @@ public class ForEachItem extends Task implements FlowableTask<VoidOutput>, Child
             this.items = items;
             this.batch = batch;
 
-            this.id = parentId + "_split";
+            this.id = parentId + SUFFIX;
             this.type = ForEachItemSplit.class.getName();
         }
 
@@ -393,6 +395,8 @@ public class ForEachItem extends Task implements FlowableTask<VoidOutput>, Child
     @Getter
     @NoArgsConstructor
     public static class ForEachItemExecutable extends Task implements ExecutableTask<Output> {
+        static final String SUFFIX = "_item";
+
         private Map<String, Object> inputs;
         private Boolean inheritLabels;
         private Map<String, String> labels;
@@ -408,7 +412,7 @@ public class ForEachItem extends Task implements FlowableTask<VoidOutput>, Child
             this.transmitFailed = transmitFailed;
             this.subflowId = subflowId;
 
-            this.id = parentId + "_executable";
+            this.id = parentId + SUFFIX;
             this.type = ForEachItemExecutable.class.getName();
         }
 
@@ -421,7 +425,7 @@ public class ForEachItem extends Task implements FlowableTask<VoidOutput>, Child
             TaskRun currentTaskRun
         ) throws InternalException {
             // get the list of splits from the outputs of the split task
-            String taskId = this.id.replace("_executable", "_split");
+            String taskId = this.id.substring(0, this.id.lastIndexOf('_')) + ForEachItemSplit.SUFFIX;
             var taskOutput = extractOutput(runContext, taskId);
             URI splitsURI = URI.create((String) taskOutput.get("splits"));
 
@@ -544,16 +548,17 @@ public class ForEachItem extends Task implements FlowableTask<VoidOutput>, Child
     @Getter
     @NoArgsConstructor
     public static class ForEachItemMergeOutputs extends Task implements RunnableTask<ForEachItemMergeOutputs.Output> {
+        static final String SUFFIX = "_merge";
 
         private ForEachItemMergeOutputs(String parentId) {
-            this.id = parentId + "_merge";
+            this.id = parentId + SUFFIX;
             this.type = ForEachItemMergeOutputs.class.getName();
         }
 
         @Override
         public ForEachItemMergeOutputs.Output run(RunContext runContext) throws Exception {
             // get the list of splits from the outputs of the split task
-            String taskId = this.id.replace("_merge", "_executable");
+            String taskId = this.id.substring(0, this.id.lastIndexOf('_')) + ForEachItemExecutable.SUFFIX;
             var taskOutput = extractOutput(runContext, taskId);
             Integer iterations = (Integer) taskOutput.get(ExecutableUtils.TASK_VARIABLE_NUMBER_OF_BATCHES);
             String subflowOutputsBaseUri = (String) taskOutput.get(ExecutableUtils.TASK_VARIABLE_SUBFLOW_OUTPUTS_BASE_URI);
diff --git a/core/src/main/java/io/kestra/core/tasks/storages/Delete.java b/core/src/main/java/io/kestra/core/tasks/storages/Delete.java
index bc08e20af..ffd2a97a7 100644
--- a/core/src/main/java/io/kestra/core/tasks/storages/Delete.java
+++ b/core/src/main/java/io/kestra/core/tasks/storages/Delete.java
@@ -8,6 +8,7 @@ import io.kestra.core.models.tasks.Task;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.storages.StorageInterface;
 import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.constraints.NotNull;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
 
@@ -37,6 +38,7 @@ public class Delete extends Task implements RunnableTask<Delete.Output> {
         description = "Must be a `kestra://` storage URI."
     )
     @PluginProperty(dynamic = true)
+    @NotNull
     private String uri;
 
     @Schema(
diff --git a/core/src/main/java/io/kestra/core/tasks/storages/Reverse.java b/core/src/main/java/io/kestra/core/tasks/storages/Reverse.java
index 4c0e73f89..478915e3a 100644
--- a/core/src/main/java/io/kestra/core/tasks/storages/Reverse.java
+++ b/core/src/main/java/io/kestra/core/tasks/storages/Reverse.java
@@ -7,6 +7,7 @@ import io.kestra.core.models.tasks.RunnableTask;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.runners.RunContext;
 import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.constraints.NotNull;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
 import org.apache.commons.io.Charsets;
@@ -43,6 +44,7 @@ public class Reverse extends Task implements RunnableTask<Reverse.Output> {
         title = "The file to be split."
     )
     @PluginProperty(dynamic = true)
+    @NotNull
     private String from;
 
     @Schema(
diff --git a/core/src/main/java/io/kestra/core/tasks/storages/Size.java b/core/src/main/java/io/kestra/core/tasks/storages/Size.java
index b3da03e5f..b372f2a23 100644
--- a/core/src/main/java/io/kestra/core/tasks/storages/Size.java
+++ b/core/src/main/java/io/kestra/core/tasks/storages/Size.java
@@ -8,6 +8,7 @@ import io.kestra.core.models.tasks.Task;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.storages.StorageInterface;
 import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.constraints.NotNull;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
 
@@ -36,6 +37,7 @@ public class Size extends Task implements RunnableTask<Size.Output> {
         description = "Must be a `kestra://` storage URI."
     )
     @PluginProperty(dynamic = true)
+    @NotNull
     private String uri;
 
     @Override
diff --git a/core/src/main/java/io/kestra/core/tasks/storages/Split.java b/core/src/main/java/io/kestra/core/tasks/storages/Split.java
index 20848a7c2..3da29b14a 100644
--- a/core/src/main/java/io/kestra/core/tasks/storages/Split.java
+++ b/core/src/main/java/io/kestra/core/tasks/storages/Split.java
@@ -10,6 +10,7 @@ import io.kestra.core.services.StorageService;
 import io.kestra.core.storages.StorageSplitInterface;
 import io.micronaut.core.convert.format.ReadableBytesTypeConverter;
 import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.constraints.NotNull;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
 
@@ -64,6 +65,7 @@ public class Split extends Task implements RunnableTask<Split.Output>, StorageSp
         title = "The file to be split."
     )
     @PluginProperty(dynamic = true)
+    @NotNull
     private String from;
 
     private String bytes;


Test patch:
diff --git a/repository-memory/src/test/resources/application-test.yml b/repository-memory/src/test/resources/application-test.yml
index 1efded100..b2f2f6c92 100644
--- a/repository-memory/src/test/resources/application-test.yml
+++ b/repository-memory/src/test/resources/application-test.yml
@@ -3,6 +3,6 @@ kestra:
     type: memory
   repository:
     type: memory
-  worker:
+  server:
     liveness:
       enabled: false
\ No newline at end of file