Repository: kestra-io/kestra (java)

Issue:
Follow-Up on Dynamic Properties -- batch size for ForEachItem cannot be passed from inputs
### Describe the issue

```
At flow: Cannot deserialize value of type `java.lang.Integer` from String "{{ inputs.batch_size}}": not a valid `java.lang.Integer` value at [Source: UNKNOWN; byte offset: #UNKNOWN] (through reference chain: io.kestra.core.models.flows.Flow["tasks"]->java.util.ArrayList[1]->io.kestra.plugin.core.flow.ForEachItem["batch"]->io.kestra.plugin.core.flow.ForEachItem$Batch["rows"])
```

Reproducer:

```yaml
id: file_output_parent_concat_foreach_batch_1
namespace: company.team
inputs:
  - id: batch_size
    type: INT
    defaults: 1

description: |
  id: file_output_child
  namespace: company.team

  inputs:
    - id: json_file
      type: FILE

  tasks:
    - id: debug
      type: io.kestra.plugin.core.log.Log
      message: "{{ json(read(inputs.json_file)) }}"

    - id: jsonata
      type: io.kestra.plugin.transform.jsonata.TransformItems
      from: "{{ inputs.json_file }}"
      expression: |
        {
          "id": id,
          "name": data.Color & ' ' & name
        }

  outputs:
    - id: myoutput
      type: FILE
      value: "{{ outputs.jsonata.uri }}"

tasks:
  - id: extract
    type: io.kestra.plugin.jdbc.duckdb.Query
    sql: |
      INSTALL httpfs;
      LOAD httpfs;
      SELECT *
      FROM read_csv_auto('https://huggingface.co/datasets/kestra/datasets/raw/main/csv/orders.csv', header=True);
    store: true

  - id: for_each_item
    type: io.kestra.plugin.core.flow.ForEachItem
    items: "{{ outputs.extract.uri }}"
    batch:
      rows: "{{ inputs.batch_size}}"
    namespace: company.team
    flowId: file_output_child
    wait: true
    transmitFailed: true
    inputs:
      order: "{{ taskrun.items }}"

  - id: ion_to_json
    type: io.kestra.plugin.serdes.json.IonToJson
    from: "{{ outputs.for_each_item_merge.subflowOutputs }}"
    newLine: false

  - id: merge
    type: io.kestra.plugin.core.storage.Concat
    files: "{{ read(outputs.ion_to_json.uri) | jq('map(.myoutput)') | flatten }}"
    separator: "\n"

  - id: for_each
    type: io.kestra.plugin.core.flow.ForEach
    values: "{{ read(outputs.ion_to_json.uri) }}"
    tasks:
      - id: log
        type: io.kestra.plugin.core.log.Log
        message: "{{ read(json(taskrun.value).myoutput) }}"
```

### Environment

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


Pull request description:
feat(core): allow dynamic properties for Split and ForEachItem
Fixes #5442

You can now define the following flow:

```yaml
id: for-each
namespace: company.team

inputs:
  - id: file
    type: FILE
  - id: batch 
    type: INT

tasks:
  - id: for-each
    type: io.kestra.plugin.core.flow.ForEachItem
    items: "{{inputs.file}}"
    batch:
      rows: "{{inputs.batch}}"
    namespace: company.team
    flowId: for-each-item
    inputs:
      file: "{{taskrun.items}}"
```


Patch:
diff --git a/core/src/main/java/io/kestra/core/runners/DefaultRunContext.java b/core/src/main/java/io/kestra/core/runners/DefaultRunContext.java
index dbce1775a..7b95e0bc8 100644
--- a/core/src/main/java/io/kestra/core/runners/DefaultRunContext.java
+++ b/core/src/main/java/io/kestra/core/runners/DefaultRunContext.java
@@ -7,6 +7,7 @@ import com.google.common.collect.ImmutableMap;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.metrics.MetricRegistry;
 import io.kestra.core.models.executions.AbstractMetricEntry;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.services.KVStoreService;
 import io.kestra.core.storages.Storage;
 import io.kestra.core.storages.StorageInterface;
@@ -204,6 +205,16 @@ public class DefaultRunContext extends RunContext {
         return variableRenderer.render(inline, mergeWithNullableValues(this.variables, variables));
     }
 
+    @Override
+    public <T> T render(Property<T> inline, Class<T> clazz) throws IllegalVariableEvaluationException {
+        return inline == null ? null : inline.as(this, clazz);
+    }
+
+    @Override
+    public <T> T render(Property<T> inline, Class<T> clazz, Map<String, Object> variables) throws IllegalVariableEvaluationException {
+        return inline == null ? null : inline.as(this, clazz, variables);
+    }
+
     /**
      * {@inheritDoc}
      */
diff --git a/core/src/main/java/io/kestra/core/runners/RunContext.java b/core/src/main/java/io/kestra/core/runners/RunContext.java
index bf1aaf8a0..748450346 100644
--- a/core/src/main/java/io/kestra/core/runners/RunContext.java
+++ b/core/src/main/java/io/kestra/core/runners/RunContext.java
@@ -6,6 +6,7 @@ import io.kestra.core.encryption.EncryptionService;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.exceptions.ResourceExpiredException;
 import io.kestra.core.models.executions.AbstractMetricEntry;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.storages.StateStore;
 import io.kestra.core.storages.Storage;
 import io.kestra.core.storages.kv.KVStore;
@@ -59,6 +60,10 @@ public abstract class RunContext {
 
     public abstract String render(String inline, Map<String, Object> variables) throws IllegalVariableEvaluationException;
 
+    public abstract <T> T render(Property<T> inline, Class<T> clazz) throws IllegalVariableEvaluationException;
+
+    public abstract <T> T  render(Property<T> inline, Class<T> clazz, Map<String, Object> variables) throws IllegalVariableEvaluationException;
+
     public abstract List<String> render(List<String> inline) throws IllegalVariableEvaluationException;
 
     public abstract List<String> render(List<String> inline, Map<String, Object> variables) throws IllegalVariableEvaluationException;
diff --git a/core/src/main/java/io/kestra/core/services/StorageService.java b/core/src/main/java/io/kestra/core/services/StorageService.java
index f7d17cf66..013b4d560 100644
--- a/core/src/main/java/io/kestra/core/services/StorageService.java
+++ b/core/src/main/java/io/kestra/core/services/StorageService.java
@@ -1,5 +1,6 @@
 package io.kestra.core.services;
 
+import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.storages.StorageSplitInterface;
 import io.micronaut.core.convert.format.ReadableBytesTypeConverter;
@@ -14,12 +15,11 @@ import java.util.List;
 import java.util.function.BiFunction;
 import java.util.stream.Collectors;
 
-import static io.kestra.core.utils.Rethrow.throwConsumer;
-import static io.kestra.core.utils.Rethrow.throwFunction;
+import static io.kestra.core.utils.Rethrow.*;
 
 public abstract class StorageService {
 
-    public static List<URI> split(RunContext runContext, StorageSplitInterface storageSplitInterface, URI from) throws IOException {
+    public static List<URI> split(RunContext runContext, StorageSplitInterface storageSplitInterface, URI from) throws IOException, IllegalVariableEvaluationException {
         String fromPath = from.getPath();
         String extension = ".tmp";
         if (fromPath.indexOf('.') >= 0) {
@@ -31,14 +31,17 @@ public abstract class StorageService {
 
             if (storageSplitInterface.getBytes() != null) {
                 ReadableBytesTypeConverter readableBytesTypeConverter = new ReadableBytesTypeConverter();
-                Number convert = readableBytesTypeConverter.convert(storageSplitInterface.getBytes(), Number.class)
+                Number convert = readableBytesTypeConverter.convert(storageSplitInterface.getBytes().as(runContext, String.class), Number.class)
                     .orElseThrow(() -> new IllegalArgumentException("Invalid size with value '" + storageSplitInterface.getBytes() + "'"));
 
-                splited = StorageService.split(runContext, extension, storageSplitInterface.getSeparator(), bufferedReader, (bytes, size) -> bytes >= convert.longValue());
+                splited = StorageService.split(runContext, extension, runContext.render(storageSplitInterface.getSeparator(), String.class),
+                    bufferedReader, (bytes, size) -> bytes >= convert.longValue());
             } else if (storageSplitInterface.getPartitions() != null) {
-                splited = StorageService.partition(runContext, extension, storageSplitInterface.getSeparator(), bufferedReader, storageSplitInterface.getPartitions());
+                splited = StorageService.partition(runContext, extension, runContext.render(storageSplitInterface.getSeparator(), String.class),
+                    bufferedReader, storageSplitInterface.getPartitions().as(runContext, Integer.class));
             } else if (storageSplitInterface.getRows() != null) {
-                splited = StorageService.split(runContext, extension, storageSplitInterface.getSeparator(), bufferedReader, (bytes, size) -> size >= storageSplitInterface.getRows());
+                splited = StorageService.split(runContext, extension, runContext.render(storageSplitInterface.getSeparator(), String.class),
+                    bufferedReader, throwBiFunction((bytes, size) -> size >= storageSplitInterface.getRows().as(runContext, Integer.class)));
             } else {
                 throw new IllegalArgumentException("Invalid configuration with no size, count, nor rows");
             }
diff --git a/core/src/main/java/io/kestra/core/storages/StorageSplitInterface.java b/core/src/main/java/io/kestra/core/storages/StorageSplitInterface.java
index 19b2a470d..e4e7c0dd3 100644
--- a/core/src/main/java/io/kestra/core/storages/StorageSplitInterface.java
+++ b/core/src/main/java/io/kestra/core/storages/StorageSplitInterface.java
@@ -1,32 +1,7 @@
 package io.kestra.core.storages;
 
-import com.google.common.base.Charsets;
-import com.google.common.hash.Hashing;
-import io.kestra.core.annotations.Retryable;
-import io.kestra.core.models.annotations.PluginProperty;
-import io.kestra.core.models.executions.Execution;
-import io.kestra.core.models.executions.TaskRun;
-import io.kestra.core.models.flows.Flow;
-import io.kestra.core.models.flows.Input;
-import io.kestra.core.models.tasks.Task;
-import io.kestra.core.models.triggers.AbstractTrigger;
-import io.kestra.core.models.triggers.TriggerContext;
-import io.kestra.core.utils.Slugify;
-import io.micronaut.core.annotation.Introspected;
-import io.micronaut.core.annotation.Nullable;
+import io.kestra.core.models.property.Property;
 import io.swagger.v3.oas.annotations.media.Schema;
-import lombok.Builder;
-
-import java.io.*;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Optional;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-import java.util.stream.Collectors;
 
 public interface StorageSplitInterface {
     @Schema(
@@ -34,25 +9,21 @@ public interface StorageSplitInterface {
         description = "Can be provided as a string in the format \"10MB\" or \"200KB\", or the number of bytes. " +
             "This allows you to process large files, slit them into smaller chunks by lines and process them in parallel. For example, MySQL by default limits the size of a query size to 16MB per query. Trying to use a bulk insert query with input data larger than 16MB will fail. Splitting the input data into smaller chunks is a common strategy to circumvent this limitation. By dividing a large data set into chunks smaller than the `max_allowed_packet` size (e.g., 10MB), you can insert the data in multiple smaller queries. This approach not only helps to avoid hitting the query size limit but can also be more efficient and manageable in terms of memory utilization, especially for very large datasets. In short, by splitting the file by bytes, you can bulk-insert smaller chunks of e.g. 10MB in parallel to avoid this limitation."
     )
-    @PluginProperty(dynamic = true)
-    String getBytes();
+    Property<String> getBytes();
 
     @Schema(
         title = "Split a file into a fixed number of partitioned files. For example, if you have a file with 1000 lines and you set `partitions` to 10, the file will be split into 10 files with 100 lines each."
     )
-    @PluginProperty(dynamic = true)
-    Integer getPartitions();
+    Property<Integer> getPartitions();
 
     @Schema(
         title = "A number of rows per batch. The file will then be split into chunks with that maximum number of rows."
     )
-    @PluginProperty(dynamic = true)
-    Integer getRows();
+    Property<Integer> getRows();
 
     @Schema(
         title = "The separator used to split a file into chunks. By default, it's a newline `\\n` character. If you are on Windows, you might want to use `\\r\\n` instead.",
         defaultValue = "\\n"
     )
-    @PluginProperty
-    String getSeparator();
+    Property<String> getSeparator();
 }
diff --git a/core/src/main/java/io/kestra/plugin/core/flow/ForEachItem.java b/core/src/main/java/io/kestra/plugin/core/flow/ForEachItem.java
index 2a3ca03fa..7d3d0387e 100644
--- a/core/src/main/java/io/kestra/plugin/core/flow/ForEachItem.java
+++ b/core/src/main/java/io/kestra/plugin/core/flow/ForEachItem.java
@@ -622,15 +622,15 @@ public class ForEachItem extends Task implements FlowableTask<VoidOutput>, Child
     @Getter
     @NoArgsConstructor
     public static class Batch implements StorageSplitInterface {
-        private String bytes;
+        private Property<String> bytes;
 
-        private Integer partitions;
+        private Property<Integer> partitions;
 
         @Builder.Default
-        private Integer rows = 1;
+        private Property<Integer> rows = Property.of(1);
 
         @Builder.Default
-        private String separator = "\n";
+        private Property<String> separator = Property.of("\n");
     }
 
     @Builder
diff --git a/core/src/main/java/io/kestra/plugin/core/storage/Split.java b/core/src/main/java/io/kestra/plugin/core/storage/Split.java
index 386b1e39b..63645e647 100644
--- a/core/src/main/java/io/kestra/plugin/core/storage/Split.java
+++ b/core/src/main/java/io/kestra/plugin/core/storage/Split.java
@@ -3,6 +3,7 @@ package io.kestra.plugin.core.storage;
 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.property.Property;
 import io.kestra.core.models.tasks.RunnableTask;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.runners.RunContext;
@@ -58,14 +59,14 @@ public class Split extends Task implements RunnableTask<Split.Output>, StorageSp
     @NotNull
     private String from;
 
-    private String bytes;
+    private Property<String> bytes;
 
-    private Integer partitions;
+    private Property<Integer> partitions;
 
-    private Integer rows;
+    private Property<Integer> rows;
 
     @Builder.Default
-    private String separator = "\n";
+    private Property<String> separator = Property.of("\n");
 
     @Override
     public Split.Output run(RunContext runContext) throws Exception {


Test patch:
diff --git a/core/src/test/java/io/kestra/plugin/core/flow/ForEachItemCaseTest.java b/core/src/test/java/io/kestra/plugin/core/flow/ForEachItemCaseTest.java
index d96fac1a5..6aa6a9dc3 100644
--- a/core/src/test/java/io/kestra/plugin/core/flow/ForEachItemCaseTest.java
+++ b/core/src/test/java/io/kestra/plugin/core/flow/ForEachItemCaseTest.java
@@ -74,7 +74,7 @@ public class ForEachItemCaseTest {
         });
 
         URI file = storageUpload();
-        Map<String, Object> inputs = Map.of("file", file.toString());
+        Map<String, Object> inputs = Map.of("file", file.toString(), "batch", 4);
         Execution execution = runnerUtils.runOne(null, TEST_NAMESPACE, "for-each-item", null,
             (flow, execution1) -> flowIO.readExecutionInputs(flow, execution1, inputs),
             Duration.ofSeconds(30));
@@ -105,7 +105,7 @@ public class ForEachItemCaseTest {
 
     public void forEachItemEmptyItems() throws TimeoutException, URISyntaxException, IOException, QueueException {
         URI file = emptyItems();
-        Map<String, Object> inputs = Map.of("file", file.toString());
+        Map<String, Object> inputs = Map.of("file", file.toString(), "batch", 4);
         Execution execution = runnerUtils.runOne(null, TEST_NAMESPACE, "for-each-item", null,
             (flow, execution1) -> flowIO.readExecutionInputs(flow, execution1, inputs),
             Duration.ofSeconds(30));
diff --git a/core/src/test/java/io/kestra/plugin/core/storage/SplitTest.java b/core/src/test/java/io/kestra/plugin/core/storage/SplitTest.java
index e1a0120c9..b79310181 100644
--- a/core/src/test/java/io/kestra/plugin/core/storage/SplitTest.java
+++ b/core/src/test/java/io/kestra/plugin/core/storage/SplitTest.java
@@ -1,6 +1,7 @@
 package io.kestra.plugin.core.storage;
 
 import com.google.common.io.CharStreams;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.runners.RunContextFactory;
 import io.kestra.core.storages.StorageInterface;
@@ -40,7 +41,7 @@ class SplitTest {
 
         Split result = Split.builder()
             .from(put.toString())
-            .partitions(8)
+            .partitions(Property.of(8))
             .build();
 
         Split.Output run = result.run(runContext);
@@ -57,7 +58,7 @@ class SplitTest {
 
         Split result = Split.builder()
             .from(put.toString())
-            .rows(10)
+            .rows(Property.of(10))
             .build();
 
         Split.Output run = result.run(runContext);
@@ -73,7 +74,7 @@ class SplitTest {
 
         Split result = Split.builder()
             .from(put.toString())
-            .bytes("1KB")
+            .bytes(Property.of("1KB"))
             .build();
 
         Split.Output run = result.run(runContext);
diff --git a/core/src/test/resources/flows/valids/for-each-item.yaml b/core/src/test/resources/flows/valids/for-each-item.yaml
index 59b2412e8..6fa5f6456 100644
--- a/core/src/test/resources/flows/valids/for-each-item.yaml
+++ b/core/src/test/resources/flows/valids/for-each-item.yaml
@@ -4,13 +4,15 @@ namespace: io.kestra.tests
 inputs:
   - id: file
     type: FILE
+  - id: batch
+    type: INT
 
 tasks:
   - id: each
     type: io.kestra.plugin.core.flow.ForEachItem
     items: "{{ inputs.file }}"
     batch:
-      rows: 4
+      rows: "{{inputs.batch}}"
     namespace: io.kestra.tests
     flowId: for-each-item-subflow
     wait: true