ForEachItem task

[loicmathieu]

Fixes #2131

What changes are being made and why?

Add a ForEachItem task that will split a file in batches and will trigger a subflow execution for each batch.

The current implementation will generates as many subflows as there is batches with unlimited concurrency.

---

How the changes have been QAed?

Parent flow

id: each-item
namespace: dev

tasks:
  - id: extract
    type: io.kestra.plugin.gcp.bigquery.Query
    sql: |
      SELECT DATETIME(datehour) as date, title, views FROM `bigquery-public-data.wikipedia.pageviews_2023` 
      WHERE DATE(datehour) = current_date() and wiki = 'fr' and title not in ('Cookie_(informatique)', 'Wikipédia:Accueil_principal', 'Spécial:Recherche')
      ORDER BY datehour desc, views desc
      LIMIT 10000
    store: true
  
  - id: each
    type: io.kestra.core.tasks.flows.ForEachItem
    items: "{{ outputs.extract.uri }}"
    maxItemsPerBatch: 10
    subflow:
      namespace: dev
      flowId: per-item
      wait: true
      transmitFailed: true

Sub flow:

id: per-item
namespace: dev

tasks:
  - id: per-item
    type: io.kestra.core.tasks.log.Log
    message: "{{ trigger.items }}"

[loicmathieu]

Note that test cannot be added to the MemoryExecutor as it fails with the known error:

2023-10-27 15:51:48,678 ERROR memory-queue_17 .TestThreadUncaughtExceptionHandlers Caught an exception in Thread[memory-queue_17,5,main]. Keep it running.
java.lang.IllegalStateException: Execution state don't exist for 1Xts6TJZEB6IRzqs15ZBtZ, receive WorkerTaskResult(taskRun=TaskRun(tenantId=null, id=7LWGpQxNqbshkyIhWjc6NP, executionId=1Xts6TJZEB6IRzqs15ZBtZ, namespace=io.kestra.tests, flowId=trigger-flow-listener-no-condition, taskId=only-listener, parentTaskRunId=null, value=null, attempts=null, outputs=null, state=State(current=RUNNING, histories=[State.History(state=CREATED, date=2023-10-27T13:51:48.541411428Z), State.History(state=RUNNING, date=2023-10-27T13:51:48.555250723Z)])), dynamicTaskRuns=null)
	at io.kestra.runner.memory.MemoryExecutor.lambda$workerTaskResultQueue$11(MemoryExecutor.java:352)
	at java.base/java.util.concurrent.ConcurrentHashMap.compute(ConcurrentHashMap.java:1916)
	at io.kestra.runner.memory.MemoryExecutor.workerTaskResultQueue(MemoryExecutor.java:350)
	at io.kestra.runner.memory.MemoryQueue.lambda$produce$0(MemoryQueue.java:79)
	at io.micrometer.core.instrument.internal.TimedRunnable.run(TimedRunnable.java:49)

I try to fix this error without success so I decided to only test the task with the JDBC and the Kafak runners.

[anna-geller]

Some minor comments on examples, otherwise awesome work! 🎸

[anna-geller]

I took it for a spin just now and it seems that the internal storage file for a given batch cannot be accessed by the subflow: https://share.descript.com/view/dLuk3uN2F0L

reproducer:

subflow

id: subflow
namespace: qa

tasks:
  - id: for_each_item
    type: io.kestra.plugin.scripts.shell.Commands
    runner: PROCESS
    commands:
      - cat "{{ trigger.items }}"

parent

id: each_parent
namespace: dev

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 }}"
    maxItemsPerBatch: 10
    subflow:
      namespace: qa
      flowId: subflow
      wait: true
      transmitFailed: true

in the same way, the preview on the Overview page doesn't work and topology view cannot be expanded for that subflow (but this last one is a UI feature that Yann can help with):

https://share.descript.com/view/EP3CKE2s5RW

[anna-geller]

see other comment, QA didn't fully pass

[loicmathieu]

@anna-geller are you sure your branch was up-to-date while doing the QA as I fixed this issue on Tuesday

[Skraye]

Or from the namespace files storage ?

[loicmathieu]

This is the standard comment, but yes, it should be possible. But note that we currently have no way to have the storage URI of a namespace file easily (we only have a Pebble function to read the content).
For now, I prefer to keep the standard comment we use everywhere

[anna-geller]

yes @loicmathieu I deleted my local branch and re-pulled again just to be sure and still exactly as described above, please try the reproducer

the server log I'm getting when trying to preview a batch file:
2023-11-02 10:04:22,096 WARN default-nioEventLoopGroup-1-13 io.kestra.webserver.access [Date: 2023-11-02T10:04:22.071922+01:00] [Duration: 24 ms] [Method: GET] [Url: /api/v1/executions/62ABgYpG6PovQxgDMtl02N/file/preview?executionId=62ABgYpG6PovQxgDMtl02N&path=kestra:%2F%2F%2Fdev%2Feach_parent%2Fexecutions%2F1xHlFEW4x3YoGngbwtEBmp%2Ftasks%2Feach%2F8dVOCngDLvdVqHdb9mEdd%2Fbatch-6.ion&maxRows=100] [Status: 404] [Length: 428] [Ip: 0:0:0:0:0:0:0:1] [Port: 8080]

[anna-geller]

I can confirm that the issue occurs only when the flow/task have underscores in the name like each_parent, without that, both accessing the internal storage file and the preview work fine

[loicmathieu]

@anna-geller this has been fixed

[sonarcloud]

Kudos, SonarCloud Quality Gate passed!   

0 Bugs
0 Vulnerabilities
0 Security Hotspots
19 Code Smells

85.9% Coverage
0.0% Duplication