Repository: kestra-io/kestra (java)

Issue:
[UI] The `ForEachItem` task is stuck in a Running state in the logs tab in the UI (even though child Executions finished)
### Feature description

this only happens in the logs tab:

![image](https://github.com/kestra-io/kestra/assets/86264395/0ed10749-0e18-4581-9419-7793c0af522d)


![image](https://github.com/kestra-io/kestra/assets/86264395/b928745d-5e19-4cdc-accc-1d30f86c6fc8)


Pull request description:
fix(core): add taskrun attemp when ending the ForEachItem task run
Fixes #2485

Patch:
diff --git a/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java b/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java
index 155657bcf..121256ea8 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java
@@ -135,8 +135,10 @@ public final class ExecutableUtils {
                     // the final state should be computed based on the iterations
                     return previousTaskRun.withOutputs(Map.of("iterations", iterations));
                 } else if (terminatedIterations == maxIterations && taskRun.getState().isTerminated()) {
+                    var state = transmitFailed ? findTerminalState(iterations) : State.Type.SUCCESS;
                     return previousTaskRun.withOutputs(Map.of("iterations", iterations))
-                        .withState(transmitFailed ? findTerminalState(iterations) : State.Type.SUCCESS);
+                        .withAttempts(Collections.singletonList(TaskRunAttempt.builder().state(new State().withState(state)).build()))
+                        .withState(state);
                 }
                 return taskRun.withOutputs(Map.of("iterations", iterations));
             }
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 cd16d8746..f5a7e26b9 100644
--- a/core/src/main/java/io/kestra/core/storages/StorageSplitInterface.java
+++ b/core/src/main/java/io/kestra/core/storages/StorageSplitInterface.java
@@ -30,27 +30,27 @@ import java.util.stream.Collectors;
 
 public interface StorageSplitInterface {
     @Schema(
-        title = "Split by file size.",
-        description = "Can be provided as a string like \"10MB\" or \"200KB\", or the number of bytes. " +
-            "Since we divide storage per line, it's not an hard requirements and files can be a larger."
+        title = "Split a large file into multiple chunks with a maximum file size of `bytes`.",
+        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();
 
     @Schema(
-        title = "Split by a fixed number of files."
+        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();
 
     @Schema(
-        title = "Split by file rows count."
+        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();
 
     @Schema(
-        title = "The separator to used between rows"
+        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."
     )
     @PluginProperty
     String getSeparator();
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 c57b2cdd8..a570a4ac8 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
@@ -50,35 +50,68 @@ import static io.kestra.core.utils.Rethrow.throwFunction;
 @Plugin(
     examples = {
         @Example(
-            title = "Execute a subflow for each batch of items",
-            code = {
+            title = """
+            Execute a subflow for each batch of items. The subflow `orders` is called from the parent flow `orders_parallel` using the `ForEachItem` task in order to start one subflow execution for each batch of items.
+            ```yaml
+            id: orders
+            namespace: prod
+
+            inputs:
+              - name: order
+                type: STRING
+
+            tasks:
+              - id: read_file
+                type: io.kestra.plugin.scripts.shell.Commands
+                runner: PROCESS
+                commands:
+                  - cat "{{ inputs.order }}"
+
+              - id: read_file_content
+                type: io.kestra.core.tasks.log.Log
+                message: "{{ read(inputs.order) }}"
+            ```
+            """,
+            full = true,
+            code =
                 """
-                    id: each
+                id: orders_parallel
+                namespace: prod
+
+                tasks:
+                  - id: extract
+                    type: io.kestra.plugin.jdbc.duckdb.Query
+                    sql: |
+                      INSTALL httpfs;
+                      LOAD httpfs;
+                      SELECT *
+                      FROM read_csv_auto('https://raw.githubusercontent.com/kestra-io/datasets/main/csv/orders.csv', header=True);
+                    store: true
+
+                  - id: each
                     type: io.kestra.core.tasks.flows.ForEachItem
-                    items: "{{ outputs.extract.uri }}" # works with API payloads too. Kestra can detect if this output is not a file,\s
-                    # and will make it to a file, split into (batches of) items
+                    items: "{{ outputs.extract.uri }}"
                     batch:
-                      rows: 10
-                    flowId: file
-                    namespace: dev
-                    inputs:
-                      file: "{{ taskrun.items }}" # special variable that contains the items of the batch
+                      rows: 1
+                    namespace: prod
+                    flowId: orders
                     wait: true # wait for the subflow execution
-                    transmitFailed: true # fail the task run if the subflow execution fails"""
-            }
+                    transmitFailed: true # fail the task run if the subflow execution fails
+                    inputs:
+                      order: "{{ taskrun.items }}" # special variable that contains the items of the batch"""
         )
     }
 )
 public class ForEachItem extends Task implements ExecutableTask<ForEachItem.Output> {
     @NotEmpty
     @PluginProperty(dynamic = true)
-    @Schema(title = "The items to be split into batches and processed. Make sure to set it to Kestra's internal storage URI, e.g. output from a previous task in the format `{{ outputs.task_id.uri }}` or an input parameter of FILE type e.g. `{{ inputs.myfile }}`.")
+    @Schema(title = "The items to be split into batches and processed. Make sure to set it to Kestra's internal storage URI. This can be either the output from a previous task, formatted as `{{ outputs.task_id.uri }}`, or a FILE type input parameter, like `{{ inputs.myfile }}`. This task is optimized for files where each line represents a single item. Suitable file types include Amazon ION-type files (commonly produced by Query tasks), newline-separated JSON files, or CSV files formatted with one row per line and without a header. For files in other formats such as Excel, CSV, Avro, Parquet, XML, or JSON, it's recommended to first convert them to the ION format. This can be done using the conversion tasks available in the `io.kestra.plugin.serdes` module, which will transform files from their original format to ION.")
     private String items;
 
     @NotNull
     @PluginProperty
     @Builder.Default
-    @Schema(title = "The batch split size")
+    @Schema(title = "How to split the items into batches.")
     private ForEachItem.Batch batch = Batch.builder().build();
 
     @NotEmpty
@@ -123,7 +156,7 @@ public class ForEachItem extends Task implements ExecutableTask<ForEachItem.Outp
 
     @Builder.Default
     @Schema(
-        title = "Whether to fail the current execution if the subflow execution fails or is killed",
+        title = "Whether to fail the current execution if the subflow execution fails or is killed.",
         description = "Note that this option works only if `wait` is set to `true`."
     )
     @PluginProperty
@@ -234,8 +267,8 @@ public class ForEachItem extends Task implements ExecutableTask<ForEachItem.Outp
     @Getter
     public static class Output implements io.kestra.core.models.tasks.Output {
         @Schema(
-            title = "The iterations counter.",
-            description = "This output will be updated in real-time with the subflow executions.\n It will contains one counter by subflow execution state, plus a `max` counter that represent the maximum number of iterations (or the number of batches)."
+            title = "The counter of iterations for each subflow execution state.",
+            description = "This output will be updated in real-time based on the state of subflow executions.\n It will contain one counter per subflow execution state, as well as a `max` counter that represents the maximum number of iterations (i.e. the total number of batches)."
         )
         private final Map<String, Integer> iterations;
     }


Test patch:
diff --git a/core/src/test/java/io/kestra/core/tasks/flows/ForEachItemCaseTest.java b/core/src/test/java/io/kestra/core/tasks/flows/ForEachItemCaseTest.java
index e6c644bb3..f4000343a 100644
--- a/core/src/test/java/io/kestra/core/tasks/flows/ForEachItemCaseTest.java
+++ b/core/src/test/java/io/kestra/core/tasks/flows/ForEachItemCaseTest.java
@@ -62,6 +62,8 @@ public class ForEachItemCaseTest {
 
         // assert on the main flow execution
         assertThat(execution.getTaskRunList(), hasSize(1));
+        assertThat(execution.getTaskRunList().get(0).getAttempts(), hasSize(1));
+        assertThat(execution.getTaskRunList().get(0).getAttempts().get(0).getState().getCurrent(), is(State.Type.SUCCESS));
         assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
         Map<String, Object> outputs = execution.getTaskRunList().get(0).getOutputs();
         assertThat(outputs.get("iterations"), notNullValue());
@@ -96,6 +98,8 @@ public class ForEachItemCaseTest {
 
         // assert on the main flow execution
         assertThat(execution.getTaskRunList(), hasSize(1));
+        assertThat(execution.getTaskRunList().get(0).getAttempts(), hasSize(1));
+        assertThat(execution.getTaskRunList().get(0).getAttempts().get(0).getState().getCurrent(), is(State.Type.SUCCESS));
         assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
         Map<String, Object> outputs = execution.getTaskRunList().get(0).getOutputs();
         assertThat(outputs.get("iterations"), notNullValue());
@@ -140,6 +144,8 @@ public class ForEachItemCaseTest {
 
         // assert on the main flow execution
         assertThat(execution.getTaskRunList(), hasSize(1));
+        assertThat(execution.getTaskRunList().get(0).getAttempts(), hasSize(1));
+        assertThat(execution.getTaskRunList().get(0).getAttempts().get(0).getState().getCurrent(), is(State.Type.FAILED));
         assertThat(execution.getState().getCurrent(), is(State.Type.FAILED));
         Map<String, Object> outputs = execution.getTaskRunList().get(0).getOutputs();
         assertThat(outputs.get("iterations"), notNullValue());