fix(executions): make state_duration generated on queries

- fixes https://github.com/kestra-io/kestra/issues/11593

.github/actions/plugins-list/action.yml

@@ -1,29 +0,0 @@
-name: 'Load Kestra Plugin List'
-description: 'Composite action to load list of plugins'
-inputs:
-  plugin-version:
-    description: "Kestra version"
-    default: 'LATEST'
-    required: true
-  plugin-file:
-    description: "File of the plugins"
-    default: './.plugins'
-    required: true
-outputs:
-  plugins:
-    description: "List of all Kestra plugins"
-    value: ${{ steps.plugins.outputs.plugins }}
-  repositories:
-    description: "List of all Kestra repositories of plugins"
-    value: ${{ steps.plugins.outputs.repositories }}
-runs:
-  using: composite
-  steps:
-    - name: Get Plugins List
-      id: plugins
-      shell: bash
-      run: |
-        PLUGINS=$([ -f ${{ inputs.plugin-file }} ] && cat ${{ inputs.plugin-file }} | grep "io\\.kestra\\." | sed -e '/#/s/^.//' | sed -e "s/LATEST/${{ inputs.plugin-version }}/g" | cut -d':' -f2- | xargs || echo '');
-        REPOSITORIES=$([ -f ${{ inputs.plugin-file }} ] && cat ${{ inputs.plugin-file }} | grep "io\\.kestra\\." | sed -e '/#/s/^.//' | cut -d':' -f1 | uniq | sort | xargs || echo '')
-        echo "plugins=$PLUGINS" >> $GITHUB_OUTPUT
-        echo "repositories=$REPOSITORIES" >> $GITHUB_OUTPUT

.github/actions/setup-vars/action.yml

@@ -1,20 +0,0 @@
-name: 'Setup vars'
-description: 'Composite action to setup common vars'
-outputs:
-  tag:
-    description: "Git tag"
-    value: ${{ steps.vars.outputs.tag }}
-  commit:
-    description: "Git commit"
-    value: ${{ steps.vars.outputs.commit }}
-runs:
-  using: composite
-  steps:
-    # Setup vars
-    - name: Set variables
-      id: vars
-      shell: bash
-      run: |
-        TAG=${GITHUB_REF#refs/*/}
-        echo "tag=${TAG}" >> $GITHUB_OUTPUT
-        echo "commit=$(git rev-parse --short "$GITHUB_SHA")" >> $GITHUB_OUTPUT

.github/workflows/e2e-scheduling.yml

@@ -0,0 +1,15 @@
+name: 'E2E tests scheduling'
+# 'New E2E tests implementation started by Roman. Based on playwright in npm UI project, tests Kestra OSS develop docker image. These tests are written from zero, lets make them unflaky from the start!.'
+on:
+  schedule:
+    - cron: "0 * * * *"  # Every hour
+  workflow_dispatch:
+    inputs:
+      noInputYet:
+        description: 'not input yet.'
+        required: false
+        type: string
+        default: "no input"
+jobs:
+  e2e:
+    uses: kestra-io/actions/.github/workflows/kestra-oss-e2e-tests.yml@main

.github/workflows/e2e.yml

@@ -1,86 +0,0 @@
-name: 'E2E tests revival'
-description: 'New E2E tests implementation started by Roman. Based on playwright in npm UI project, tests Kestra OSS develop docker image. These tests are written from zero, lets make them unflaky from the start!.'
-on:
-  schedule:
-    - cron: "0 * * * *"  # Every hour
-  workflow_call:
-    inputs:
-      noInputYet:
-        description: 'not input yet.'
-        required: false
-        type: string
-        default: "no input"
-  workflow_dispatch:
-    inputs:
-      noInputYet:
-        description: 'not input yet.'
-        required: false
-        type: string
-        default: "no input"
-jobs:
-  check:
-    timeout-minutes: 15
-    runs-on: ubuntu-latest
-    env:
-      GOOGLE_SERVICE_ACCOUNT: ${{ secrets.GOOGLE_SERVICE_ACCOUNT }}
-    steps:
-      - name: Login to DockerHub
-        uses: docker/login-action@v3
-        with:
-          registry: ghcr.io
-          username: ${{ github.actor }}
-          password: ${{ github.token }}
-
-      - name: Checkout kestra
-        uses: actions/checkout@v5
-        with:
-          path: kestra
-
-      # Setup build
-      - uses: kestra-io/actions/composite/setup-build@main
-        name: Setup - Build
-        id: build
-        with:
-          java-enabled: true
-          node-enabled: true
-          python-enabled: true
-
-      - name: Install Npm dependencies
-        run: |
-          cd kestra/ui
-          npm i
-          npx playwright install --with-deps chromium
-
-      - name: Run E2E Tests
-        run: |
-          cd kestra
-          sh build-and-start-e2e-tests.sh
-
-      - name: Upload Playwright Report as Github artifact
-        # 'With this report, you can analyze locally the results of the tests. see https://playwright.dev/docs/ci-intro#html-report'
-        uses: actions/upload-artifact@v4
-        if: ${{ !cancelled() }}
-        with:
-          name: playwright-report
-          path: kestra/ui/playwright-report/
-          retention-days: 7
-        # Allure check
-        # TODO I don't know what it should do
-#      - uses: rlespinasse/github-slug-action@v5
-#        name: Allure - Generate slug variables
-#
-#      - name: Allure - Publish report
-#        uses: andrcuns/allure-publish-action@v2.9.0
-#        if: always() && env.GOOGLE_SERVICE_ACCOUNT != ''
-#        continue-on-error: true
-#        env:
-#          GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_AUTH_TOKEN }}
-#          JAVA_HOME: /usr/lib/jvm/default-jvm/
-#        with:
-#          storageType: gcs
-#          resultsGlob: "**/build/allure-results"
-#          bucket: internal-kestra-host
-#          baseUrl: "https://internal.dev.kestra.io"
-#          prefix: ${{ format('{0}/{1}', github.repository, 'allure/java') }}
-#          copyLatest: true
-#          ignoreMissingResults: true

.github/workflows/global-create-new-release-branch.yml

@@ -1,5 +1,5 @@
-name: Run Gradle Release
-run-name: "Releasing Kestra ${{ github.event.inputs.releaseVersion }} 🚀"
+name: Create new release branch
+run-name: "Create new release branch Kestra ${{ github.event.inputs.releaseVersion }} 🚀"
 on:
   workflow_dispatch:
     inputs:

.github/workflows/global-gradle-release-plugins.yml

@@ -35,7 +35,7 @@ jobs:
 
       # Get Plugins List
       - name: Get Plugins List
-        uses: ./.github/actions/plugins-list
+        uses: kestra-io/actions/composite/kestra-oss/kestra-oss-plugins-list@main
         id: plugins-list
         with:
           plugin-version: 'LATEST'

.github/workflows/global-setversion-tag-plugins.yml

@@ -23,7 +23,7 @@ jobs:
 
       # Get Plugins List
       - name: Get Plugins List
-        uses: ./.github/actions/plugins-list
+        uses: kestra-io/actions/composite/kestra-oss/kestra-oss-plugins-list@main
         id: plugins-list
         with:
           plugin-version: 'LATEST'
@@ -40,7 +40,7 @@ jobs:
           GITHUB_PAT: ${{ secrets.GH_PERSONAL_TOKEN }}
         run: |
           chmod +x ./dev-tools/setversion-tag-plugins.sh;
-          
+
           ./dev-tools/setversion-tag-plugins.sh \
           --release-version=${{github.event.inputs.releaseVersion}} \
           --yes \
@@ -52,7 +52,7 @@ jobs:
           GITHUB_PAT: ${{ secrets.GH_PERSONAL_TOKEN }}
         run: |
           chmod +x ./dev-tools/setversion-tag-plugins.sh;
-          
+
           ./dev-tools/setversion-tag-plugins.sh \
           --release-version=${{github.event.inputs.releaseVersion}} \
           --dry-run \

.github/workflows/global-start-release.yml

@@ -1,5 +1,5 @@
-name: Set Version and Tag
-run-name: "Set version and Tag Kestra to ${{ github.event.inputs.releaseVersion }} 🚀"
+name: Start release
+run-name: "Start release of Kestra ${{ github.event.inputs.releaseVersion }} 🚀"
 on:
   workflow_dispatch:
     inputs:
@@ -7,17 +7,26 @@ on:
         description: 'The release version (e.g., 0.21.1)'
         required: true
         type: string
+
+permissions:
+  contents: write
+
 env:
   RELEASE_VERSION: "${{ github.event.inputs.releaseVersion }}"
 jobs:
   release:
     name: Release Kestra
     runs-on: ubuntu-latest
-    if: startsWith(github.ref, 'refs/heads/releases/v')
     steps:
-      # Checks
-      - name: Check Inputs
+      - name: Parse and Check Inputs
+        id: parse-and-check-inputs
         run: |
+          CURRENT_BRANCH="${{ github.ref_name }}"
+          if ! [[ "$CURRENT_BRANCH" == "develop" ]]; then
+            echo "You can only run this workflow on develop, but you ran it on $CURRENT_BRANCH"
+            exit 1
+          fi
+
           if ! [[ "$RELEASE_VERSION" =~ ^[0-9]+(\.[0-9]+)(\.[0-9]+)(-rc[0-9])?(-SNAPSHOT)?$ ]]; then
             echo "Invalid release version. Must match regex: ^[0-9]+(\.[0-9]+)(\.[0-9]+)-(rc[0-9])?(-SNAPSHOT)?$"
             exit 1
@@ -25,13 +34,8 @@ jobs:
 
           # Extract the major and minor versions
           BASE_VERSION=$(echo "$RELEASE_VERSION" | sed -E 's/^([0-9]+\.[0-9]+)\..*/\1/')
-          RELEASE_BRANCH="refs/heads/releases/v${BASE_VERSION}.x"
-
-          CURRENT_BRANCH="$GITHUB_REF"
-          if ! [[ "$CURRENT_BRANCH" == "$RELEASE_BRANCH" ]]; then
-            echo "Invalid release branch. Expected $RELEASE_BRANCH, was $CURRENT_BRANCH"
-            exit 1
-          fi
+          RELEASE_BRANCH="releases/v${BASE_VERSION}.x"
+          echo "release_branch=${RELEASE_BRANCH}" >> $GITHUB_OUTPUT
 
       # Checkout
       - name: Checkout
@@ -39,6 +43,7 @@ jobs:
         with:
           fetch-depth: 0
           token: ${{ secrets.GH_PERSONAL_TOKEN }}
+          ref: ${{ steps.parse-and-check-inputs.outputs.release_branch }}
 
       # Configure
       - name: Git - Configure
@@ -47,7 +52,7 @@ jobs:
           git config --global user.name "github-actions[bot]"
 
       # Execute
-      - name: Run Gradle Release
+      - name: Start release by updating version and pushing a new tag
         env:
           GITHUB_PAT: ${{ secrets.GH_PERSONAL_TOKEN }}
         run: |

.github/workflows/main-build.yml

@@ -1,6 +1,11 @@
 name: Main Workflow
 
 on:
+  push:
+    branches:
+      - releases/*
+      - develop
+
   workflow_dispatch:
     inputs:
       skip-test:
@@ -11,53 +16,58 @@ on:
         options:
           - "true"
           - "false"
-      plugin-version:
-          description: "plugins version"
-          required: false
-          type: string
-  push:
-    branches:
-      - master
-      - main
-      - releases/*
-      - develop
-    tags:
-      - v*
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.ref }}-main
   cancel-in-progress: true
 
 jobs:
-  tests:
-    name: Execute tests
-    uses: ./.github/workflows/workflow-test.yml
+  backend-tests:
+    name: Backend tests
     if: ${{ github.event.inputs.skip-test == 'false' || github.event.inputs.skip-test == '' }}
-    with:
-      report-status: false
+    uses: kestra-io/actions/.github/workflows/kestra-oss-backend-tests.yml@main
+    secrets:
+      GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+      CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+      SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
+      GOOGLE_SERVICE_ACCOUNT: ${{ secrets.GOOGLE_SERVICE_ACCOUNT }}
 
-  release:
-    name: Release
-    needs: [tests]
-    if: "!failure() && !cancelled() && !startsWith(github.ref, 'refs/heads/releases')"
-    uses: ./.github/workflows/workflow-release.yml
+  frontend-tests:
+    name: Frontend tests
+    if: ${{ github.event.inputs.skip-test == 'false' || github.event.inputs.skip-test == '' }}
+    uses: kestra-io/actions/.github/workflows/kestra-oss-frontend-tests.yml@main
+    secrets:
+      CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+      GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+
+  publish-develop-docker:
+    name: Publish Docker
+    needs: [backend-tests, frontend-tests]
+    if: "!failure() && !cancelled() && github.ref == 'refs/heads/develop'"
+    uses: kestra-io/actions/.github/workflows/kestra-oss-publish-docker.yml@main
     with:
-        plugin-version: ${{ inputs.plugin-version != '' && inputs.plugin-version || (github.ref == 'refs/heads/develop' && 'LATEST-SNAPSHOT' || 'LATEST') }}
+      plugin-version: 'LATEST-SNAPSHOT'
     secrets:
       DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
       DOCKERHUB_PASSWORD: ${{ secrets.DOCKERHUB_PASSWORD }}
+      SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
+
+
+  publish-develop-maven:
+    name: Publish develop Maven
+    needs: [ backend-tests, frontend-tests ]
+    if: "!failure() && !cancelled() && github.ref == 'refs/heads/develop'"
+    uses: kestra-io/actions/.github/workflows/kestra-oss-publish-maven.yml@main
+    secrets:
       SONATYPE_USER: ${{ secrets.SONATYPE_USER }}
       SONATYPE_PASSWORD: ${{ secrets.SONATYPE_PASSWORD }}
       SONATYPE_GPG_KEYID: ${{ secrets.SONATYPE_GPG_KEYID }}
       SONATYPE_GPG_PASSWORD: ${{ secrets.SONATYPE_GPG_PASSWORD }}
       SONATYPE_GPG_FILE: ${{ secrets.SONATYPE_GPG_FILE }}
-      GH_PERSONAL_TOKEN: ${{ secrets.GH_PERSONAL_TOKEN }}
-      SLACK_RELEASES_WEBHOOK_URL: ${{ secrets.SLACK_RELEASES_WEBHOOK_URL }}
 
   end:
     runs-on: ubuntu-latest
-    needs:
-      - release
+    needs: [publish-develop-docker, publish-develop-maven]
     if: always()
     steps:
       - name: Trigger EE Workflow

.github/workflows/pre-release.yml

@@ -0,0 +1,49 @@
+name: Pre Release
+
+on:
+  push:
+    tags:
+      - 'v*'
+  workflow_dispatch:
+
+jobs:
+  build-artifacts:
+    name: Build Artifacts
+    uses: kestra-io/actions/.github/workflows/kestra-oss-build-artifacts.yml@main
+
+  backend-tests:
+    name: Backend tests
+    uses: kestra-io/actions/.github/workflows/kestra-oss-backend-tests.yml@main
+    secrets:
+      GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+      CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+      SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
+      GOOGLE_SERVICE_ACCOUNT: ${{ secrets.GOOGLE_SERVICE_ACCOUNT }}
+
+  frontend-tests:
+    name: Frontend tests
+    uses: kestra-io/actions/.github/workflows/kestra-oss-frontend-tests.yml@main
+    secrets:
+      GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+      CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+
+  publish-maven:
+    name: Publish Maven
+    needs: [ backend-tests, frontend-tests ]
+    if: "!failure() && !cancelled()"
+    uses: kestra-io/actions/.github/workflows/kestra-oss-publish-maven.yml@main
+    secrets:
+      SONATYPE_USER: ${{ secrets.SONATYPE_USER }}
+      SONATYPE_PASSWORD: ${{ secrets.SONATYPE_PASSWORD }}
+      SONATYPE_GPG_KEYID: ${{ secrets.SONATYPE_GPG_KEYID }}
+      SONATYPE_GPG_PASSWORD: ${{ secrets.SONATYPE_GPG_PASSWORD }}
+      SONATYPE_GPG_FILE: ${{ secrets.SONATYPE_GPG_FILE }}
+
+  publish-github:
+    name: Github Release
+    needs: [build-artifacts, backend-tests, frontend-tests]
+    if: "!failure() && !cancelled()"
+    uses: kestra-io/actions/.github/workflows/kestra-oss-publish-github.yml@main
+    secrets:
+      GH_PERSONAL_TOKEN: ${{ secrets.GH_PERSONAL_TOKEN }}
+      SLACK_RELEASES_WEBHOOK_URL: ${{ secrets.SLACK_RELEASES_WEBHOOK_URL }}

.github/workflows/pull-request-cleanup.yml

@@ -3,7 +3,7 @@ name: Pull Request - Delete Docker
 on:
   pull_request:
     types: [closed]
-
+# TODO import a reusable one
 jobs:
   publish:
     name: Pull Request - Delete Docker

.github/workflows/pull-request.yml

@@ -2,17 +2,12 @@ name: Pull Request Workflow
 
 on:
   pull_request:
-    branches:
-      - develop
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.ref_name }}-pr
   cancel-in-progress: true
 
 jobs:
-  # ********************************************************************************************************************
-  # File changes detection
-  # ********************************************************************************************************************
   file-changes:
     if: ${{ github.event.pull_request.draft == false }}
     name: File changes detection
@@ -33,14 +28,11 @@ jobs:
               - '!{ui,.github}/**'
           token: ${{ secrets.GITHUB_TOKEN }}
 
-  # ********************************************************************************************************************
-  # Tests
-  # ********************************************************************************************************************
   frontend:
     name: Frontend - Tests
     needs: [file-changes]
     if: "needs.file-changes.outputs.ui == 'true'"
-    uses: ./.github/workflows/workflow-frontend-test.yml
+    uses: kestra-io/actions/.github/workflows/kestra-oss-frontend-tests.yml@main
     secrets:
       GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
       CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
@@ -49,7 +41,7 @@ jobs:
     name: Backend - Tests
     needs: file-changes
     if: "needs.file-changes.outputs.backend == 'true'"
-    uses: ./.github/workflows/workflow-backend-test.yml
+    uses: kestra-io/actions/.github/workflows/kestra-oss-backend-tests.yml@main
     secrets:
       GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
       CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
@@ -58,5 +50,8 @@ jobs:
 
   e2e-tests:
     name: E2E - Tests
-    uses: ./.github/workflows/e2e.yml
+    uses: kestra-io/actions/.github/workflows/kestra-oss-e2e-tests.yml@main
 
+  generate-pull-request-docker-image:
+    name: Generate PR docker image
+    uses: kestra-io/actions/.github/workflows/kestra-oss-pullrequest-publish-docker.yml@main

.github/workflows/release-docker.yml

@@ -0,0 +1,34 @@
+name: Publish docker
+
+on:
+  workflow_dispatch:
+    inputs:
+      retag-latest:
+        description: 'Retag latest Docker images'
+        required: true
+        type: boolean
+        default: false
+      retag-lts:
+        description: 'Retag LTS Docker images'
+        required: true
+        type: boolean
+        default: false
+      plugin-version:
+        description: 'Plugin version'
+        required: false
+        type: string
+        default: "LATEST"
+
+jobs:
+  publish-docker:
+    name: Publish Docker
+    if: startsWith(github.ref, 'refs/tags/v')
+    uses: kestra-io/actions/.github/workflows/kestra-oss-publish-docker.yml@main
+    with:
+      plugin-version: ${{ inputs.plugin-version }}
+      retag-latest: ${{ inputs.retag-latest }}
+      retag-lts: ${{ inputs.retag-lts }}
+    secrets:
+      DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
+      DOCKERHUB_PASSWORD: ${{ secrets.DOCKERHUB_PASSWORD }}
+      SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}

.github/workflows/workflow-backend-test.yml

@@ -1,74 +0,0 @@
-name: Backend - Tests
-
-on:
-  workflow_call:
-    secrets:
-      GITHUB_AUTH_TOKEN:
-        description: "The GitHub Token."
-        required: true
-      CODECOV_TOKEN:
-        description: 'Codecov Token'
-        required: true
-      SONAR_TOKEN:
-        description: 'Sonar Token'
-        required: true
-      GOOGLE_SERVICE_ACCOUNT:
-        description: 'Google Service Account'
-        required: true
-
-permissions:
-  contents: write
-  checks: write
-  actions: read
-  pull-requests: write
-
-jobs:
-  test:
-    name: Backend - Tests
-    runs-on: ubuntu-latest
-    env:
-      GOOGLE_SERVICE_ACCOUNT: ${{ secrets.GOOGLE_SERVICE_ACCOUNT }}
-      SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
-    steps:
-      - uses: actions/checkout@v5
-        name: Checkout - Current ref
-        with:
-          fetch-depth: 0
-
-      # Setup build
-      - uses: kestra-io/actions/composite/setup-build@main
-        name: Setup - Build
-        id: build
-        with:
-          java-enabled: true
-          node-enabled: true
-          python-enabled: true
-
-      # Services
-      - name: Setup - Start docker compose
-        shell: bash
-        run: docker compose -f docker-compose-ci.yml up -d
-
-      # Gradle check
-      - name: Gradle - Build
-        if: ${{ github.event.inputs.skip-test == 'false' || github.event.inputs.skip-test == '' }}
-        env:
-          GOOGLE_SERVICE_ACCOUNT: ${{ secrets.GOOGLE_SERVICE_ACCOUNT }}
-        shell: bash
-        run: |
-          echo $GOOGLE_SERVICE_ACCOUNT | base64 -d > ~/.gcp-service-account.json
-          export GOOGLE_APPLICATION_CREDENTIALS=$HOME/.gcp-service-account.json
-          ./gradlew check javadoc --parallel
-
-      - name: comment PR with test report
-        if: ${{ !cancelled() && github.event_name == 'pull_request' }}
-        env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_AUTH_TOKEN }}
-        run: npx --yes @kestra-io/kestra-devtools generateTestReportSummary --only-errors --ci $(pwd)
-
-      # Report Java
-      - name: Report - Java
-        uses: kestra-io/actions/composite/report-java@main
-        if: ${{ !cancelled() }}
-        with:
-          secrets: ${{ toJSON(secrets) }}

.github/workflows/workflow-build-artifacts.yml

@@ -1,80 +0,0 @@
-name: Build Artifacts
-
-on:
-  workflow_call: {}
-
-jobs:
-  build:
-    name: Build - Artifacts
-    runs-on: ubuntu-latest
-    outputs:
-        docker-tag: ${{ steps.vars.outputs.tag }}
-        docker-artifact-name: ${{ steps.vars.outputs.artifact }}
-        plugins: ${{ steps.plugins.outputs.plugins }}
-    env:
-      PLUGIN_VERSION: ${{ github.event.inputs.plugin-version != null && github.event.inputs.plugin-version || 'LATEST' }}
-    steps:
-      - name: Checkout - Current ref
-        uses: actions/checkout@v5
-        with:
-          fetch-depth: 0
-
-      # Npm
-      - name: Setup - Npm install
-        shell: bash
-        working-directory: ui
-        run: npm ci
-
-      # Setup build
-      - uses: kestra-io/actions/composite/setup-build@main
-        name: Setup - Build
-        id: build
-        with:
-          java-enabled: true
-          node-enabled: true
-
-      # Get Plugins List
-      - name: Plugins - Get List
-        uses: ./.github/actions/plugins-list
-        if: "!startsWith(github.ref, 'refs/tags/v')"
-        id: plugins-list
-        with:
-          plugin-version: ${{ env.PLUGIN_VERSION }}
-
-      # Set Plugins List
-      - name: Plugins - Set List
-        id: plugins
-        if: "!startsWith(github.ref, 'refs/tags/v')"
-        shell: bash
-        run: |
-          PLUGINS="${{ steps.plugins-list.outputs.plugins }}"
-          TAG=${GITHUB_REF#refs/*/}
-          if [[ $TAG = "master" || $TAG == v* ]]; then
-            echo "plugins=$PLUGINS" >> $GITHUB_OUTPUT
-          else
-            echo "plugins=--repositories=https://central.sonatype.com/repository/maven-snapshots/ $PLUGINS" >> $GITHUB_OUTPUT
-          fi
-
-      # Build
-      - name: Gradle - Build
-        shell: bash
-        run: |
-          ./gradlew executableJar
-
-      - name: Artifacts - Copy exe to image
-        shell: bash
-        run: |
-          cp build/executable/* docker/app/kestra && chmod +x docker/app/kestra
-
-      # Upload artifacts
-      - name: Artifacts - Upload JAR
-        uses: actions/upload-artifact@v4
-        with:
-          name: jar
-          path: build/libs/
-
-      - name: Artifacts - Upload Executable
-        uses: actions/upload-artifact@v4
-        with:
-          name: exe
-          path: build/executable/

.github/workflows/workflow-frontend-test.yml

@@ -1,70 +0,0 @@
-name: Frontend - Tests
-
-on:
-  workflow_call:
-    secrets:
-      GITHUB_AUTH_TOKEN:
-        description: "The GitHub Token."
-        required: true
-      CODECOV_TOKEN:
-        description: 'Codecov Token'
-        required: true
-
-env:
-  # to save corepack from itself
-  COREPACK_INTEGRITY_KEYS: 0
-
-jobs:
-  test:
-    name: Frontend - Tests
-    runs-on: ubuntu-latest
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v5
-
-      - name: Cache Node Modules
-        id: cache-node-modules
-        uses: actions/cache@v4
-        with:
-          path: |
-            ui/node_modules
-          key: modules-${{ hashFiles('ui/package-lock.json') }}
-
-      - name: Cache Playwright Binaries
-        id: cache-playwright
-        uses: actions/cache@v4
-        with:
-          path: |
-            ~/.cache/ms-playwright
-          key: playwright-${{ hashFiles('ui/package-lock.json') }}
-
-      - name: Npm - install
-        if: steps.cache-node-modules.outputs.cache-hit != 'true'
-        working-directory: ui
-        run: npm ci
-
-      - name: Npm - lint
-        uses: reviewdog/action-eslint@v1
-        with:
-          github_token: ${{ secrets.GITHUB_AUTH_TOKEN }}
-          reporter: github-pr-review
-          workdir: ui
-
-      - name: Npm - Run build
-        working-directory: ui
-        env:
-          CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
-        run: npm run build
-
-      - name: Run front-end unit tests
-        working-directory: ui
-        run: npm run test:unit -- --coverage
-
-      - name: Storybook - Install Playwright
-        working-directory: ui
-        if: steps.cache-playwright.outputs.cache-hit != 'true'
-        run: npx playwright install --with-deps
-
-      - name: Run storybook component tests
-        working-directory: ui
-        run: npm run test:storybook -- --coverage

.github/workflows/workflow-github-release.yml

@@ -1,79 +0,0 @@
-name: Github - Release
-
-on:
-  workflow_dispatch:
-  workflow_call:
-    secrets:
-      GH_PERSONAL_TOKEN:
-        description: "The Github personal token."
-        required: true
-      SLACK_RELEASES_WEBHOOK_URL:
-        description: "The Slack webhook URL."
-        required: true
-
-
-
-jobs:
-  publish:
-    name: Github - Release
-    runs-on: ubuntu-latest
-    steps:
-      # Check out
-      - name: Checkout - Repository
-        uses: actions/checkout@v5
-        with:
-          fetch-depth: 0
-          submodules: true
-
-
-      # Download Exec
-      # Must be done after checkout actions
-      - name: Artifacts - Download executable
-        uses: actions/download-artifact@v5
-        if: startsWith(github.ref, 'refs/tags/v')
-        with:
-          name: exe
-          path: build/executable
-
-      - name: Check if current tag is latest
-        id: is_latest
-        run: |
-          latest_tag=$(git tag | grep -E '^v[0-9]+\.[0-9]+\.[0-9]+$' | sed 's/^v//' | sort -V | tail -n1)
-          current_tag="${GITHUB_REF_NAME#v}"
-          if [ "$current_tag" = "$latest_tag" ]; then
-          echo "latest=true" >> $GITHUB_OUTPUT
-          else
-          echo "latest=false" >> $GITHUB_OUTPUT
-          fi
-        env:
-          GITHUB_REF_NAME: ${{ github.ref_name }}
-
-      # GitHub Release
-      - name: Create GitHub release
-        uses: kestra-io/actions/composite/github-release@main
-        if: ${{ startsWith(github.ref, 'refs/tags/v') }}
-        env:
-          MAKE_LATEST: ${{ steps.is_latest.outputs.latest }}
-          GITHUB_TOKEN: ${{ secrets.GH_PERSONAL_TOKEN }}
-          SLACK_RELEASES_WEBHOOK_URL: ${{ secrets.SLACK_RELEASES_WEBHOOK_URL }}
-
-      # Trigger gha workflow to bump helm chart version
-      - name: GitHub - Trigger the Helm chart version bump
-        uses: peter-evans/repository-dispatch@v3
-        with:
-          token: ${{ secrets.GH_PERSONAL_TOKEN }}
-          repository: kestra-io/helm-charts
-          event-type: update-helm-chart-version
-          client-payload: |-
-            {
-              "new_version": "${{ github.ref_name }}",
-              "github_repository": "${{ github.repository }}",
-              "github_actor": "${{ github.actor }}"
-            }
-
-      - name: Merge Release Notes
-        if: ${{ startsWith(github.ref, 'refs/tags/v') }}
-        uses: kestra-io/actions/composite/github-release-note-merge@main
-        env:
-            GITHUB_TOKEN: ${{ secrets.GH_PERSONAL_TOKEN }}
-            RELEASE_TAG: ${{ github.ref_name }}

.github/workflows/workflow-publish-docker.yml

@@ -1,208 +0,0 @@
-name: Create Docker images on Release
-
-on:
-  workflow_dispatch:
-    inputs:
-      retag-latest:
-        description: 'Retag latest Docker images'
-        required: true
-        type: choice
-        default: "false"
-        options:
-          - "true"
-          - "false"
-      retag-lts:
-        description: 'Retag LTS Docker images'
-        required: true
-        type: choice
-        default: "false"
-        options:
-          - "true"
-          - "false"
-      release-tag:
-        description: 'Kestra Release Tag (by default, deduced with the ref)'
-        required: false
-        type: string
-      plugin-version:
-        description: 'Plugin version'
-        required: false
-        type: string
-        default: "LATEST"
-      force-download-artifact:
-        description: 'Force download artifact'
-        required: false
-        type: choice
-        default: "true"
-        options:
-          - "true"
-          - "false"
-  workflow_call:
-    inputs:
-      plugin-version:
-        description: "Plugin version"
-        default: 'LATEST'
-        required: false
-        type: string
-      force-download-artifact:
-        description: 'Force download artifact'
-        required: false
-        type: string
-        default: "true"
-    secrets:
-      DOCKERHUB_USERNAME:
-        description: "The Dockerhub username."
-        required: true
-      DOCKERHUB_PASSWORD:
-        description: "The Dockerhub password."
-        required: true
-
-env:
-  PLUGIN_VERSION: ${{ inputs.plugin-version != null && inputs.plugin-version || 'LATEST' }}
-jobs:
-  plugins:
-    name: List Plugins
-    runs-on: ubuntu-latest
-    outputs:
-      plugins: ${{ steps.plugins.outputs.plugins }}
-    steps:
-      # Checkout
-      - uses: actions/checkout@v5
-
-      # Get Plugins List
-      - name: Get Plugins List
-        uses: ./.github/actions/plugins-list
-        id: plugins
-        with: # remap LATEST-SNAPSHOT to LATEST
-          plugin-version: ${{ env.PLUGIN_VERSION == 'LATEST-SNAPSHOT' && 'LATEST' || env.PLUGIN_VERSION }}
-
-  # ********************************************************************************************************************
-  # Build
-  # ********************************************************************************************************************
-  build-artifacts:
-    name: Build Artifacts
-    if: ${{ inputs.force-download-artifact == 'true' }}
-    uses: ./.github/workflows/workflow-build-artifacts.yml
-
-  docker:
-    name: Publish Docker
-    needs: [ plugins, build-artifacts ]
-    if: always()
-    runs-on: ubuntu-latest
-    strategy:
-      matrix:
-        image:
-          - name: "-no-plugins"
-            plugins: ""
-            packages: jattach
-            python-libs: ""
-          - name: ""
-            plugins: ${{needs.plugins.outputs.plugins}}
-            packages: python3 python-is-python3 python3-pip curl jattach
-            python-libs: kestra
-    steps:
-      - uses: actions/checkout@v5
-
-      # Vars
-      - name: Set image name
-        id: vars
-        run: |
-          if [[ "${{ inputs.release-tag }}" == "" ]]; then
-            TAG=${GITHUB_REF#refs/*/}
-            echo "tag=${TAG}" >> $GITHUB_OUTPUT
-          else
-            TAG="${{ inputs.release-tag }}"
-            echo "tag=${TAG}" >> $GITHUB_OUTPUT
-          fi
-
-          if [[ $GITHUB_REF == refs/tags/* ]]; then
-            if [[ $TAG =~ ^v[0-9]+\.[0-9]+\.[0-9]+$ ]]; then
-              # this will remove the patch version number
-              MINOR_SEMVER=${TAG%.*}
-              echo "minor_semver=${MINOR_SEMVER}" >> $GITHUB_OUTPUT
-            else
-              echo "Tag '$TAG' is not a valid semver (vMAJOR.MINOR.PATCH), skipping minor_semver"
-            fi
-          fi
-
-          if [[ "${{ env.PLUGIN_VERSION }}" == *"-SNAPSHOT" ]]; then
-            echo "plugins=--repositories=https://central.sonatype.com/repository/maven-snapshots/ ${{ matrix.image.plugins }}" >> $GITHUB_OUTPUT;
-          else
-            echo "plugins=${{ matrix.image.plugins }}" >> $GITHUB_OUTPUT
-          fi
-
-      # Download executable from artifact
-      - name: Artifacts - Download executable
-        uses: actions/download-artifact@v5
-        with:
-          name: exe
-          path: build/executable
-
-      - name: Copy exe to image
-        run: |
-          cp build/executable/* docker/app/kestra && chmod +x docker/app/kestra
-
-      # Docker setup
-      - name: Set up QEMU
-        uses: docker/setup-qemu-action@v3
-
-      - name: Docker - Fix Qemu
-        shell: bash
-        run: |
-          docker run --rm --privileged multiarch/qemu-user-static --reset -p yes -c yes
-
-      - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v3
-
-      # Docker Login
-      - name: Login to DockerHub
-        uses: docker/login-action@v3
-        with:
-          username: ${{ secrets.DOCKERHUB_USERNAME }}
-          password: ${{ secrets.DOCKERHUB_PASSWORD }}
-
-      # Docker Build and push
-      - name: Push to Docker Hub
-        uses: docker/build-push-action@v6
-        with:
-          context: .
-          push: true
-          tags: ${{ format('kestra/kestra:{0}{1}', steps.vars.outputs.tag, matrix.image.name) }}
-          platforms: linux/amd64,linux/arm64
-          build-args: |
-            KESTRA_PLUGINS=${{ steps.vars.outputs.plugins }}
-            APT_PACKAGES=${{ matrix.image.packages }}
-            PYTHON_LIBRARIES=${{ matrix.image.python-libs }}
-
-      - name: Install regctl
-        if: startsWith(github.ref, 'refs/tags/v')
-        uses: regclient/actions/regctl-installer@main
-
-      - name: Retag to minor semver version
-        if: startsWith(github.ref, 'refs/tags/v') && steps.vars.outputs.minor_semver != ''
-        run: |
-          regctl image copy ${{ format('kestra/kestra:{0}{1}', steps.vars.outputs.tag, matrix.image.name) }} ${{ format('kestra/kestra:{0}{1}', steps.vars.outputs.minor_semver, matrix.image.name) }}
-
-      - name: Retag to latest
-        if: startsWith(github.ref, 'refs/tags/v') && inputs.retag-latest == 'true'
-        run: |
-          regctl image copy ${{ format('kestra/kestra:{0}{1}', steps.vars.outputs.tag, matrix.image.name) }} ${{ format('kestra/kestra:latest{0}', matrix.image.name) }}
-
-      - name: Retag to LTS
-        if: startsWith(github.ref, 'refs/tags/v') && inputs.retag-lts == 'true'
-        run: |
-          regctl image copy ${{ format('kestra/kestra:{0}{1}', steps.vars.outputs.tag, matrix.image.name) }} ${{ format('kestra/kestra:latest-lts{0}', matrix.image.name) }}
-
-  end:
-    runs-on: ubuntu-latest
-    needs:
-      - docker
-    if: always()
-    env:
-      SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
-    steps:
-      - name: Slack notification
-        if: ${{ failure() && env.SLACK_WEBHOOK_URL != 0 }}
-        uses: kestra-io/actions/composite/slack-status@main
-        with:
-          webhook-url: ${{ secrets.SLACK_WEBHOOK_URL }}
-

.github/workflows/workflow-publish-maven.yml

@@ -1,57 +0,0 @@
-name: Publish - Maven
-
-on:
-  workflow_call:
-    secrets:
-      SONATYPE_USER:
-        description: "The Sonatype username."
-        required: true
-      SONATYPE_PASSWORD:
-        description: "The Sonatype password."
-        required: true
-      SONATYPE_GPG_KEYID:
-        description: "The Sonatype GPG key id."
-        required: true
-      SONATYPE_GPG_PASSWORD:
-        description: "The Sonatype GPG password."
-        required: true
-      SONATYPE_GPG_FILE:
-        description: "The Sonatype GPG file."
-        required: true
-
-jobs:
-  publish:
-    name: Publish - Maven
-    runs-on: ubuntu-latest
-    steps:
-      - name: Checkout - Current ref
-        uses: actions/checkout@v5
-
-      # Setup build
-      - name: Setup - Build
-        uses: kestra-io/actions/composite/setup-build@main
-        id: build
-        with:
-          java-enabled: true
-          node-enabled: true
-
-      # Publish
-      - name: Publish - Release package to Maven Central
-        shell: bash
-        env:
-          ORG_GRADLE_PROJECT_mavenCentralUsername: ${{ secrets.SONATYPE_USER }}
-          ORG_GRADLE_PROJECT_mavenCentralPassword: ${{ secrets.SONATYPE_PASSWORD }}
-          SONATYPE_GPG_KEYID: ${{ secrets.SONATYPE_GPG_KEYID }}
-          SONATYPE_GPG_PASSWORD: ${{ secrets.SONATYPE_GPG_PASSWORD }}
-          SONATYPE_GPG_FILE: ${{ secrets.SONATYPE_GPG_FILE}}
-        run: |
-          mkdir -p ~/.gradle/
-          echo "signing.keyId=${SONATYPE_GPG_KEYID}" > ~/.gradle/gradle.properties
-          echo "signing.password=${SONATYPE_GPG_PASSWORD}" >> ~/.gradle/gradle.properties
-          echo "signing.secretKeyRingFile=${HOME}/.gradle/secring.gpg" >> ~/.gradle/gradle.properties
-          echo ${SONATYPE_GPG_FILE} | base64 -d > ~/.gradle/secring.gpg
-          ./gradlew publishToMavenCentral
-
-      # Gradle dependency
-      - name: Java - Gradle dependency graph
-        uses: gradle/actions/dependency-submission@v4

.github/workflows/workflow-pullrequest-publish-docker.yml

@@ -1,78 +0,0 @@
-name: Pull Request - Publish Docker
-
-on:
-  pull_request:
-    branches:
-      - develop
-
-jobs:
-  build-artifacts:
-    name: Build Artifacts
-    if: github.repository == 'kestra-io/kestra' # prevent running on forks
-    uses: ./.github/workflows/workflow-build-artifacts.yml
-
-  publish:
-    name: Publish Docker
-    if: github.repository == 'kestra-io/kestra' # prevent running on forks
-    runs-on: ubuntu-latest
-    needs: build-artifacts
-    env:
-      GITHUB_IMAGE_PATH: "ghcr.io/kestra-io/kestra-pr"
-    steps:
-      - name: Checkout - Current ref
-        uses: actions/checkout@v5
-        with:
-          fetch-depth: 0
-
-      # Docker setup
-      - name: Docker - Setup QEMU
-        uses: docker/setup-qemu-action@v3
-
-      - name: Docker - Setup Docker Buildx
-        uses: docker/setup-buildx-action@v3
-
-      # Docker Login
-      - name: Login to GHCR
-        uses: docker/login-action@v3
-        with:
-          registry: ghcr.io
-          username: ${{ github.actor }}
-          password: ${{ secrets.GITHUB_TOKEN }}
-
-      # Build Docker Image
-      - name: Artifacts - Download executable
-        uses: actions/download-artifact@v5
-        with:
-          name: exe
-          path: build/executable
-
-      - name: Docker - Copy exe to image
-        shell: bash
-        run: |
-          cp build/executable/* docker/app/kestra && chmod +x docker/app/kestra
-
-      - name: Docker - Build image
-        uses: docker/build-push-action@v6
-        with:
-          context: .
-          file: ./Dockerfile.pr
-          push: true
-          tags: ${{ env.GITHUB_IMAGE_PATH }}:${{ github.event.pull_request.number }}
-          platforms: linux/amd64,linux/arm64
-
-      # Add comment on pull request
-      - name: Add comment to PR
-        uses: actions/github-script@v8
-        with:
-          github-token: ${{ secrets.GITHUB_TOKEN }}
-          script: |
-            await github.rest.issues.createComment({
-              issue_number: context.issue.number,
-              owner: context.repo.owner,
-              repo: context.repo.repo,
-              body: `**🐋 Docker image**: \`${{ env.GITHUB_IMAGE_PATH }}:${{ github.event.pull_request.number }}\`\n` +
-                `\n` +
-                `\`\`\`bash\n` +
-                `docker run --pull=always --rm -it -p 8080:8080 --user=root -v /var/run/docker.sock:/var/run/docker.sock -v /tmp:/tmp ${{ env.GITHUB_IMAGE_PATH }}:${{ github.event.pull_request.number }} server local\n` +
-                `\`\`\``
-            })

.github/workflows/workflow-release.yml

@@ -1,85 +0,0 @@
-name: Release
-
-on:
-  workflow_dispatch:
-    inputs:
-      plugin-version:
-        description: "plugins version"
-        default: 'LATEST'
-        required: false
-        type: string
-      publish-docker:
-        description: "Publish Docker image"
-        default: 'false'
-        required: false
-        type: string
-  workflow_call:
-    inputs:
-      plugin-version:
-        description: "plugins version"
-        default: 'LATEST'
-        required: false
-        type: string
-    secrets:
-      DOCKERHUB_USERNAME:
-        description: "The Dockerhub username."
-        required: true
-      DOCKERHUB_PASSWORD:
-        description: "The Dockerhub password."
-        required: true
-      SONATYPE_USER:
-        description: "The Sonatype username."
-        required: true
-      SONATYPE_PASSWORD:
-        description: "The Sonatype password."
-        required: true
-      SONATYPE_GPG_KEYID:
-        description: "The Sonatype GPG key id."
-        required: true
-      SONATYPE_GPG_PASSWORD:
-        description: "The Sonatype GPG password."
-        required: true
-      SONATYPE_GPG_FILE:
-        description: "The Sonatype GPG file."
-        required: true
-      GH_PERSONAL_TOKEN:
-        description: "GH personnal Token."
-        required: true
-      SLACK_RELEASES_WEBHOOK_URL:
-        description: "Slack webhook for releases channel."
-        required: true
-jobs:
-  build-artifacts:
-    name: Build - Artifacts
-    uses: ./.github/workflows/workflow-build-artifacts.yml
-
-  Docker:
-    name: Publish Docker
-    needs: build-artifacts
-    uses: ./.github/workflows/workflow-publish-docker.yml
-    if: github.ref == 'refs/heads/develop' || inputs.publish-docker == 'true'
-    with:
-      force-download-artifact: 'false'
-      plugin-version: ${{ inputs.plugin-version != null && inputs.plugin-version || 'LATEST' }}
-    secrets:
-      DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
-      DOCKERHUB_PASSWORD: ${{ secrets.DOCKERHUB_PASSWORD }}
-
-  Maven:
-    name: Publish Maven
-    uses: ./.github/workflows/workflow-publish-maven.yml
-    secrets:
-      SONATYPE_USER: ${{ secrets.SONATYPE_USER }}
-      SONATYPE_PASSWORD: ${{ secrets.SONATYPE_PASSWORD }}
-      SONATYPE_GPG_KEYID: ${{ secrets.SONATYPE_GPG_KEYID }}
-      SONATYPE_GPG_PASSWORD: ${{ secrets.SONATYPE_GPG_PASSWORD }}
-      SONATYPE_GPG_FILE: ${{ secrets.SONATYPE_GPG_FILE }}
-
-  Github:
-    name: Github Release
-    needs: build-artifacts
-    if: startsWith(github.ref, 'refs/tags/v')
-    uses: ./.github/workflows/workflow-github-release.yml
-    secrets:
-      GH_PERSONAL_TOKEN: ${{ secrets.GH_PERSONAL_TOKEN }}
-      SLACK_RELEASES_WEBHOOK_URL: ${{ secrets.SLACK_RELEASES_WEBHOOK_URL }}

.github/workflows/workflow-test.yml

@@ -1,95 +0,0 @@
-name: Tests
-
-on:
-  schedule:
-    - cron: '0 4 * * 1,2,3,4,5'
-  workflow_call:
-    inputs:
-      report-status:
-        description: "Report status of the jobs in outputs"
-        type: string
-        required: false
-        default: false
-    outputs:
-      frontend_status:
-        description: "Status of the frontend job"
-        value: ${{ jobs.set-frontend-status.outputs.frontend_status }}
-      backend_status:
-        description: "Status of the backend job"
-        value: ${{ jobs.set-backend-status.outputs.backend_status }}
-
-jobs:
-  file-changes:
-    name: File changes detection
-    runs-on: ubuntu-latest
-    timeout-minutes: 60
-    outputs:
-      ui: ${{ steps.changes.outputs.ui }}
-      backend: ${{ steps.changes.outputs.backend }}
-    steps:
-      - uses: actions/checkout@v5
-        if: "!startsWith(github.ref, 'refs/tags/v')"
-      - uses: dorny/paths-filter@v3
-        if: "!startsWith(github.ref, 'refs/tags/v')"
-        id: changes
-        with:
-          filters: |
-            ui:
-              - 'ui/**'
-            backend:
-              - '!{ui,.github}/**'
-          token: ${{ secrets.GITHUB_TOKEN }}
-
-  frontend:
-    name: Frontend - Tests
-    needs: file-changes
-    if: "needs.file-changes.outputs.ui == 'true' || startsWith(github.ref, 'refs/tags/v')"
-    uses: ./.github/workflows/workflow-frontend-test.yml
-    secrets:
-      GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-      CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
-
-
-  backend:
-    name: Backend - Tests
-    needs: file-changes
-    if: "needs.file-changes.outputs.backend == 'true' || startsWith(github.ref, 'refs/tags/v')"
-    uses: ./.github/workflows/workflow-backend-test.yml
-    secrets:
-      GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-      CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
-      SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
-      GOOGLE_SERVICE_ACCOUNT: ${{ secrets.GOOGLE_SERVICE_ACCOUNT }}
-
-  # Output every job status
-  # To be used in other workflows
-  report-status:
-    name: Report Status
-    runs-on: ubuntu-latest
-    needs: [ frontend, backend ]
-    if: always() && (inputs.report-status == 'true')
-    outputs:
-      frontend_status: ${{ steps.set-frontend-status.outputs.frontend_status }}
-      backend_status: ${{ steps.set-backend-status.outputs.backend_status }}
-    steps:
-      - id: set-frontend-status
-        name: Set frontend job status
-        run: echo "::set-output name=frontend_status::${{ needs.frontend.result }}"
-
-      - id: set-backend-status
-        name: Set backend job status
-        run: echo "::set-output name=backend_status::${{ needs.backend.result }}"
-
-  notify:
-    name: Notify - Slack
-    runs-on: ubuntu-latest
-    needs: [ frontend, backend ]
-    steps:
-      - name: Notify failed CI
-        if: |
-          always() &&
-            (needs.frontend.result != 'success' || needs.backend.result != 'success') &&
-            (github.ref == 'refs/heads/master' || github.ref == 'refs/heads/main' || github.ref == 'refs/heads/develop')
-        uses: kestra-io/actions/composite/slack-status@main
-        with:
-          webhook-url: ${{ secrets.SLACK_WEBHOOK_URL }}

README.md

@@ -19,9 +19,12 @@
 <br />
 
 <p align="center">
-    <a href="https://x.com/kestra_io"><img height="25" src="https://kestra.io/twitter.svg" alt="X(formerly Twitter)" /></a> &nbsp;
-    <a href="https://www.linkedin.com/company/kestra/"><img height="25" src="https://kestra.io/linkedin.svg" alt="linkedin" /></a> &nbsp;
-<a href="https://www.youtube.com/@kestra-io"><img height="25" src="https://kestra.io/youtube.svg" alt="youtube" /></a> &nbsp;
+  <a href="https://twitter.com/kestra_io" style="margin: 0 10px;">
+        <img height="25" src="https://kestra.io/twitter.svg" alt="twitter" width="35" height="25" /></a>
+  <a href="https://www.linkedin.com/company/kestra/" style="margin: 0 10px;">
+        <img height="25" src="https://kestra.io/linkedin.svg" alt="linkedin" width="35" height="25" /></a> 
+  <a href="https://www.youtube.com/@kestra-io" style="margin: 0 10px;">
+        <img height="25" src="https://kestra.io/youtube.svg" alt="youtube" width="35" height="25" /></a>
 </p>
 
 <p align="center">

build.gradle

@@ -25,7 +25,7 @@ plugins {
     id 'jacoco-report-aggregation'
 
     // helper
-    id "com.github.ben-manes.versions" version "0.52.0"
+    id "com.github.ben-manes.versions" version "0.53.0"
 
     // front
     id 'com.github.node-gradle.node' version '7.1.0'
@@ -37,7 +37,7 @@ plugins {
     id "com.vanniktech.maven.publish" version "0.34.0"
 
     // OWASP dependency check
-    id "org.owasp.dependencycheck" version "12.1.3" apply false
+    id "org.owasp.dependencycheck" version "12.1.5" apply false
 }
 
 idea {

cli/src/main/java/io/kestra/cli/commands/servers/AbstractServerCommand.java

@@ -2,19 +2,27 @@ package io.kestra.cli.commands.servers;
 
 import io.kestra.cli.AbstractCommand;
 import io.kestra.core.contexts.KestraContext;
-import jakarta.annotation.PostConstruct;
+import lombok.extern.slf4j.Slf4j;
 import picocli.CommandLine;
 
-abstract public class AbstractServerCommand extends AbstractCommand implements ServerCommandInterface {
+@Slf4j
+public abstract class AbstractServerCommand extends AbstractCommand implements ServerCommandInterface {
     @CommandLine.Option(names = {"--port"}, description = "The port to bind")
     Integer serverPort;
 
     @Override
     public Integer call()  throws Exception {
+        log.info("Machine information: {} available cpu(s), {}MB max memory, Java version {}", Runtime.getRuntime().availableProcessors(), maxMemoryInMB(), Runtime.version());
+
         this.shutdownHook(true, () -> KestraContext.getContext().shutdown());
+
         return super.call();
     }
 
+    private long maxMemoryInMB() {
+        return Runtime.getRuntime().maxMemory() / 1024 / 1024;
+    }
+
     protected static int defaultWorkerThread() {
         return Runtime.getRuntime().availableProcessors() * 8;
     }

cli/src/main/java/io/kestra/cli/commands/servers/ExecutorCommand.java

@@ -3,7 +3,7 @@ package io.kestra.cli.commands.servers;
 import com.google.common.collect.ImmutableMap;
 import io.kestra.core.models.ServerType;
 import io.kestra.core.runners.ExecutorInterface;
-import io.kestra.executor.SkipExecutionService;
+import io.kestra.core.services.SkipExecutionService;
 import io.kestra.core.services.StartExecutorService;
 import io.kestra.core.utils.Await;
 import io.micronaut.context.ApplicationContext;

cli/src/main/java/io/kestra/cli/commands/servers/IndexerCommand.java

@@ -4,10 +4,13 @@ import com.google.common.collect.ImmutableMap;
 import io.kestra.core.models.ServerType;
 import io.kestra.core.runners.Indexer;
 import io.kestra.core.utils.Await;
+import io.kestra.core.services.SkipExecutionService;
 import io.micronaut.context.ApplicationContext;
 import jakarta.inject.Inject;
 import picocli.CommandLine;
 
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 
 @CommandLine.Command(
@@ -17,6 +20,11 @@ import java.util.Map;
 public class IndexerCommand extends AbstractServerCommand {
     @Inject
     private ApplicationContext applicationContext;
+    @Inject
+    private SkipExecutionService skipExecutionService;
+
+    @CommandLine.Option(names = {"--skip-indexer-records"}, split=",", description = "a list of indexer record keys, separated by a coma; for troubleshooting purpose only")
+    private List<String> skipIndexerRecords = Collections.emptyList();
 
     @SuppressWarnings("unused")
     public static Map<String, Object> propertiesOverrides() {
@@ -27,6 +35,8 @@ public class IndexerCommand extends AbstractServerCommand {
 
     @Override
     public Integer call() throws Exception {
+        this.skipExecutionService.setSkipIndexerRecords(skipIndexerRecords);
+
         super.call();
 
         Indexer indexer = applicationContext.getBean(Indexer.class);

cli/src/main/java/io/kestra/cli/commands/servers/StandAloneCommand.java

@@ -7,7 +7,7 @@ import io.kestra.core.contexts.KestraContext;
 import io.kestra.core.models.ServerType;
 import io.kestra.core.repositories.LocalFlowRepositoryLoader;
 import io.kestra.cli.StandAloneRunner;
-import io.kestra.executor.SkipExecutionService;
+import io.kestra.core.services.SkipExecutionService;
 import io.kestra.core.services.StartExecutorService;
 import io.kestra.core.utils.Await;
 import io.micronaut.context.ApplicationContext;
@@ -63,6 +63,9 @@ public class StandAloneCommand extends AbstractServerCommand {
     @CommandLine.Option(names = {"--skip-tenants"}, split=",", description = "a list of tenants to skip, separated by a coma; for troubleshooting purpose only")
     private List<String> skipTenants = Collections.emptyList();
 
+    @CommandLine.Option(names = {"--skip-indexer-records"}, split=",", description = "a list of indexer record keys, separated by a coma; for troubleshooting purpose only")
+    private List<String> skipIndexerRecords = Collections.emptyList();
+
     @CommandLine.Option(names = {"--no-tutorials"}, description = "Flag to disable auto-loading of tutorial flows.")
     boolean tutorialsDisabled = false;
 
@@ -93,6 +96,7 @@ public class StandAloneCommand extends AbstractServerCommand {
         this.skipExecutionService.setSkipFlows(skipFlows);
         this.skipExecutionService.setSkipNamespaces(skipNamespaces);
         this.skipExecutionService.setSkipTenants(skipTenants);
+        this.skipExecutionService.setSkipIndexerRecords(skipIndexerRecords);
         this.startExecutorService.applyOptions(startExecutors, notStartExecutors);
 
         KestraContext.getContext().injectWorkerConfigs(workerThread, null);

cli/src/main/java/io/kestra/cli/commands/servers/WebServerCommand.java

@@ -5,12 +5,15 @@ import io.kestra.core.models.ServerType;
 import io.kestra.core.runners.Indexer;
 import io.kestra.core.utils.Await;
 import io.kestra.core.utils.ExecutorsUtils;
+import io.kestra.core.services.SkipExecutionService;
 import io.micronaut.context.ApplicationContext;
 import jakarta.inject.Inject;
 import lombok.extern.slf4j.Slf4j;
 import picocli.CommandLine;
 import picocli.CommandLine.Option;
 
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 
@@ -28,11 +31,17 @@ public class WebServerCommand extends AbstractServerCommand {
     @Inject
     private ExecutorsUtils executorsUtils;
 
+    @Inject
+    private SkipExecutionService skipExecutionService;
+
     @Option(names = {"--no-tutorials"}, description = "Flag to disable auto-loading of tutorial flows.")
-    boolean tutorialsDisabled = false;
+    private boolean tutorialsDisabled = false;
 
     @Option(names = {"--no-indexer"}, description = "Flag to disable starting an embedded indexer.")
-    boolean indexerDisabled = false;
+    private boolean indexerDisabled = false;
+
+    @CommandLine.Option(names = {"--skip-indexer-records"}, split=",", description = "a list of indexer record keys, separated by a coma; for troubleshooting purpose only")
+    private List<String> skipIndexerRecords = Collections.emptyList();
 
     @Override
     public boolean isFlowAutoLoadEnabled() {
@@ -48,6 +57,8 @@ public class WebServerCommand extends AbstractServerCommand {
 
     @Override
     public Integer call() throws Exception {
+        this.skipExecutionService.setSkipIndexerRecords(skipIndexerRecords);
+
         super.call();
 
         // start the indexer

cli/src/main/resources/application.yml

@@ -169,6 +169,7 @@ kestra:
         - "/api/v1/executions/webhook/"
         - "/api/v1/main/executions/webhook/"
         - "/api/v1/*/executions/webhook/"
+        - "/api/v1/basicAuthValidationErrors"
 
     preview:
       initial-rows: 100

cli/src/test/java/io/kestra/cli/services/FileChangedEventListenerTest.java

@@ -1,5 +1,6 @@
 package io.kestra.cli.services;
 
+import io.kestra.core.junit.annotations.FlakyTest;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.GenericFlow;
 import io.kestra.core.repositories.FlowRepositoryInterface;
@@ -18,6 +19,7 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import org.junitpioneer.jupiter.RetryingTest;
 
 import static io.kestra.core.utils.Rethrow.throwRunnable;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -56,7 +58,8 @@ class FileChangedEventListenerTest {
         }
     }
 
-    @Test
+    @FlakyTest
+    @RetryingTest(2)
     void test() throws IOException, TimeoutException {
         var tenant = TestsUtils.randomTenant(FileChangedEventListenerTest.class.getSimpleName(), "test");
         // remove the flow if it already exists
@@ -94,7 +97,8 @@ class FileChangedEventListenerTest {
         );
     }
 
-    @Test
+    @FlakyTest
+    @RetryingTest(2)
     void testWithPluginDefault() throws IOException, TimeoutException {
         var tenant = TestsUtils.randomTenant(FileChangedEventListenerTest.class.getName(), "testWithPluginDefault");
         // remove the flow if it already exists

core/build.gradle

@@ -84,7 +84,7 @@ dependencies {
 
     testImplementation "org.testcontainers:testcontainers:1.21.3"
     testImplementation "org.testcontainers:junit-jupiter:1.21.3"
-    testImplementation "org.bouncycastle:bcpkix-jdk18on:1.81"
+    testImplementation "org.bouncycastle:bcpkix-jdk18on"
 
     testImplementation "org.wiremock:wiremock-jetty12"
 }

core/src/main/java/io/kestra/core/docs/JsonSchemaGenerator.java

@@ -118,7 +118,7 @@ public class JsonSchemaGenerator {
             removeRequiredOnPropsWithDefaults(objectNode);
 
             return MAPPER.convertValue(objectNode, MAP_TYPE_REFERENCE);
-        } catch (IllegalArgumentException e) {
+        } catch (Exception e) {
             throw new IllegalArgumentException("Unable to generate jsonschema for '" + cls.getName() + "'", e);
         }
     }

core/src/main/java/io/kestra/core/lineage/DataSet.java

@@ -1,24 +0,0 @@
-package io.kestra.core.lineage;
-
-import io.kestra.core.models.DeletedInterface;
-import io.kestra.core.models.HasUID;
-import io.kestra.core.models.TenantInterface;
-import io.kestra.core.utils.IdUtils;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.experimental.SuperBuilder;
-
-@Getter
-@SuperBuilder(toBuilder = true)
-@NoArgsConstructor
-public class DataSet implements HasUID, TenantInterface, DeletedInterface {
-    private String tenantId;
-    private String namespace;
-    private String name;
-    private boolean deleted;
-
-    @Override
-    public String uid() {
-        return IdUtils.fromParts(tenantId, namespace, name);
-    }
-}

core/src/main/java/io/kestra/core/lineage/events/DataSetEvent.java

@@ -1,20 +0,0 @@
-package io.kestra.core.lineage.events;
-
-import io.kestra.core.lineage.DataSet;
-
-public record DataSetEvent (DataSet dataSet, LineageEventType eventType) implements LineageEvent {
-    @Override
-    public String getTenantId() {
-        return dataSet.getTenantId();
-    }
-
-    @Override
-    public String uid() {
-        return dataSet.uid();
-    }
-
-    @Override
-    public String getType() {
-        return this.getClass().getName();
-    }
-}

core/src/main/java/io/kestra/core/lineage/events/JobEvent.java

@@ -1,20 +0,0 @@
-package io.kestra.core.lineage.events;
-
-import io.kestra.core.models.flows.FlowInterface;
-
-public record JobEvent (FlowInterface flow, LineageEventType eventType) implements LineageEvent {
-    @Override
-    public String getTenantId() {
-        return flow.getTenantId();
-    }
-
-    @Override
-    public String uid() {
-        return flow.uid();
-    }
-
-    @Override
-    public String getType() {
-        return this.getClass().getName();
-    }
-}

core/src/main/java/io/kestra/core/lineage/events/LineageEvent.java

@@ -1,10 +0,0 @@
-package io.kestra.core.lineage.events;
-
-import com.fasterxml.jackson.annotation.JsonTypeInfo;
-import io.kestra.core.models.HasUID;
-import io.kestra.core.models.TenantInterface;
-
-@JsonTypeInfo(use = JsonTypeInfo.Id.CLASS, property = "type", visible = true, include = JsonTypeInfo.As.EXISTING_PROPERTY)
-public sealed interface LineageEvent extends TenantInterface, HasUID permits DataSetEvent, JobEvent, RunEvent {
-    String getType();
-}

core/src/main/java/io/kestra/core/lineage/events/LineageEventType.java

@@ -1,7 +0,0 @@
-package io.kestra.core.lineage.events;
-
-public enum LineageEventType {
-    CREATE,
-    UPDATE,
-    DELETE,
-}

core/src/main/java/io/kestra/core/lineage/events/RunEvent.java

@@ -1,20 +0,0 @@
-package io.kestra.core.lineage.events;
-
-import io.kestra.core.models.executions.Execution;
-
-public record RunEvent(Execution execution, LineageEventType eventType) implements LineageEvent {
-    @Override
-    public String getTenantId() {
-        return execution.getTenantId();
-    }
-
-    @Override
-    public String uid() {
-        return execution.getId();
-    }
-
-    @Override
-    public String getType() {
-        return this.getClass().getName();
-    }
-}

core/src/main/java/io/kestra/core/models/Label.java

@@ -1,13 +1,16 @@
 package io.kestra.core.models;
 
 import io.kestra.core.utils.MapUtils;
+import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.annotation.Nullable;
-import jakarta.validation.constraints.NotNull;
+import jakarta.validation.constraints.NotEmpty;
 
 import java.util.*;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
-public record Label(@NotNull String key, @NotNull String value) {
+@Schema(description = "A key/value pair that can be attached to a Flow or Execution. Labels are often used to organize and categorize objects.")
+public record Label(@NotEmpty String key, @NotEmpty String value) {
     public static final String SYSTEM_PREFIX = "system.";
 
     // system labels
@@ -41,7 +44,7 @@ public record Label(@NotNull String key, @NotNull String value) {
     public static Map<String, String> toMap(@Nullable List<Label> labels) {
         if (labels == null || labels.isEmpty()) return Collections.emptyMap();
         return labels.stream()
-            .filter(label -> label.value() != null && label.key() != null)
+            .filter(label -> label.value() != null && !label.value().isEmpty() && label.key() != null && !label.key().isEmpty())
             // using an accumulator in case labels with the same key exists: the second is kept
             .collect(Collectors.toMap(Label::key, Label::value, (first, second) -> second, LinkedHashMap::new));
     }
@@ -56,6 +59,7 @@ public record Label(@NotNull String key, @NotNull String value) {
     public static List<Label> deduplicate(@Nullable List<Label> labels) {
         if (labels == null || labels.isEmpty()) return Collections.emptyList();
         return toMap(labels).entrySet().stream()
+            .filter(getEntryNotEmptyPredicate())
             .map(entry -> new Label(entry.getKey(), entry.getValue()))
             .collect(Collectors.toCollection(ArrayList::new));
     }
@@ -70,6 +74,7 @@ public record Label(@NotNull String key, @NotNull String value) {
         if (map == null || map.isEmpty()) return List.of();
         return map.entrySet()
             .stream()
+            .filter(getEntryNotEmptyPredicate())
             .map(entry -> new Label(entry.getKey(), entry.getValue()))
             .toList();
     }
@@ -88,4 +93,14 @@ public record Label(@NotNull String key, @NotNull String value) {
         }
         return map;
     }
+
+    /**
+     * Provides predicate for not empty entries.
+     *
+     * @return The non-empty filter
+     */
+    public static Predicate<Map.Entry<String, String>> getEntryNotEmptyPredicate() {
+        return entry -> entry.getKey() != null && !entry.getKey().isEmpty() &&
+            entry.getValue() != null && !entry.getValue().isEmpty();
+    }
 }

core/src/main/java/io/kestra/core/models/QueryFilter.java

@@ -254,19 +254,7 @@ public record QueryFilter(
          *
          * @return List of {@code ResourceField} with resource names, fields, and operations.
          */
-        public static List<ResourceField> asResourceList() {
-            return Arrays.stream(values())
-                .map(Resource::toResourceField)
-                .toList();
-        }
-
-        private static ResourceField toResourceField(Resource resource) {
-            List<FieldOp> fieldOps = resource.supportedField().stream()
-                .map(Resource::toFieldInfo)
-                .toList();
-            return new ResourceField(resource.name().toLowerCase(), fieldOps);
-        }
-
+        
         private static FieldOp toFieldInfo(Field field) {
             List<Operation> operations = field.supportedOp().stream()
                 .map(Resource::toOperation)
@@ -279,9 +267,6 @@ public record QueryFilter(
         }
     }
 
-    public record ResourceField(String name, List<FieldOp> fields) {
-    }
-
     public record FieldOp(String name, String value, List<Operation> operations) {
     }
 

core/src/main/java/io/kestra/core/models/collectors/ExecutionUsage.java

@@ -17,31 +17,12 @@ import java.util.List;
 @Introspected
 public class ExecutionUsage {
     private final List<DailyExecutionStatistics> dailyExecutionsCount;
-    private final List<DailyExecutionStatistics> dailyTaskRunsCount;
 
     public static ExecutionUsage of(final String tenantId,
                                     final ExecutionRepositoryInterface executionRepository,
                                     final ZonedDateTime from,
                                     final ZonedDateTime to) {
 
-        List<DailyExecutionStatistics> dailyTaskRunsCount = null;
-
-        try {
-            dailyTaskRunsCount = executionRepository.dailyStatistics(
-                null,
-                tenantId,
-                null,
-                null,
-                null,
-                from,
-                to,
-                DateUtils.GroupType.DAY,
-                null,
-                true);
-        } catch (UnsupportedOperationException ignored) {
-
-        }
-
         return ExecutionUsage.builder()
             .dailyExecutionsCount(executionRepository.dailyStatistics(
                 null,
@@ -52,28 +33,13 @@ public class ExecutionUsage {
                 from,
                 to,
                 DateUtils.GroupType.DAY,
-                null,
-                false))
-            .dailyTaskRunsCount(dailyTaskRunsCount)
+                null))
             .build();
     }
 
     public static ExecutionUsage of(final ExecutionRepositoryInterface repository,
                                     final ZonedDateTime from,
                                     final ZonedDateTime to) {
-        List<DailyExecutionStatistics> dailyTaskRunsCount = null;
-        try {
-            dailyTaskRunsCount = repository.dailyStatisticsForAllTenants(
-                null,
-                null,
-                null,
-                from,
-                to,
-                DateUtils.GroupType.DAY,
-                true
-            );
-        } catch (UnsupportedOperationException ignored) {}
-
         return ExecutionUsage.builder()
             .dailyExecutionsCount(repository.dailyStatisticsForAllTenants(
                 null,
@@ -81,10 +47,8 @@ public class ExecutionUsage {
                 null,
                 from,
                 to,
-                DateUtils.GroupType.DAY,
-                false
+                DateUtils.GroupType.DAY
             ))
-            .dailyTaskRunsCount(dailyTaskRunsCount)
             .build();
     }
 }

core/src/main/java/io/kestra/core/models/executions/Execution.java

@@ -272,7 +272,7 @@ public class Execution implements DeletedInterface, TenantInterface {
     }
 
     public Execution withTaskRun(TaskRun taskRun) throws InternalException {
-        ArrayList<TaskRun> newTaskRunList = new ArrayList<>(this.taskRunList);
+        ArrayList<TaskRun> newTaskRunList = this.taskRunList == null ? new ArrayList<>() : new ArrayList<>(this.taskRunList);
 
         boolean b = Collections.replaceAll(
             newTaskRunList,
@@ -865,20 +865,18 @@ public class Execution implements DeletedInterface, TenantInterface {
      * @param e the exception raise
      * @return new taskRun with updated attempt with logs
      */
-    private FailedTaskRunWithLog lastAttemptsTaskRunForFailedExecution(TaskRun taskRun,
-        TaskRunAttempt lastAttempt, Exception e) {
+    private FailedTaskRunWithLog lastAttemptsTaskRunForFailedExecution(TaskRun taskRun, TaskRunAttempt lastAttempt, Exception e) {
+        TaskRun failed = taskRun
+            .withAttempts(
+                Stream
+                    .concat(
+                        taskRun.getAttempts().stream().limit(taskRun.getAttempts().size() - 1),
+                        Stream.of(lastAttempt.getState().isFailed() ? lastAttempt : lastAttempt.withState(State.Type.FAILED))
+                    )
+                    .toList()
+            );
         return new FailedTaskRunWithLog(
-            taskRun
-                .withAttempts(
-                    Stream
-                        .concat(
-                            taskRun.getAttempts().stream().limit(taskRun.getAttempts().size() - 1),
-                            Stream.of(lastAttempt
-                                .withState(State.Type.FAILED))
-                        )
-                        .toList()
-                )
-                .withState(State.Type.FAILED),
+            failed.getState().isFailed() ? failed : failed.withState(State.Type.FAILED),
             RunContextLogger.logEntries(loggingEventFromException(e), LogEntry.of(taskRun, kind))
         );
     }

core/src/main/java/io/kestra/core/models/executions/TaskRun.java

@@ -296,7 +296,7 @@ public class TaskRun implements TenantInterface {
     }
 
     public TaskRun incrementIteration() {
-        int iteration = this.iteration == null ? 1 : this.iteration;
+        int iteration = this.iteration == null ? 0 : this.iteration;
         return this.toBuilder()
             .iteration(iteration + 1)
             .build();

core/src/main/java/io/kestra/core/models/flows/AbstractFlow.java

@@ -3,7 +3,6 @@ package io.kestra.core.models.flows;
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import com.fasterxml.jackson.databind.annotation.JsonSerialize;
 import io.kestra.core.models.Label;
-import io.kestra.core.models.annotations.PluginProperty;
 import io.kestra.core.models.tasks.WorkerGroup;
 import io.kestra.core.serializers.ListOrMapOfLabelDeserializer;
 import io.kestra.core.serializers.ListOrMapOfLabelSerializer;
@@ -61,7 +60,14 @@ public abstract class AbstractFlow implements FlowInterface {
 
     @JsonSerialize(using = ListOrMapOfLabelSerializer.class)
     @JsonDeserialize(using = ListOrMapOfLabelDeserializer.class)
-    @Schema(implementation = Object.class, oneOf = {List.class, Map.class})
+    @Schema(
+            description = "Labels as a list of Label (key/value pairs) or as a map of string to string.",
+            oneOf = {
+                    Label[].class,
+                    Map.class
+            }
+    )
+    @Valid
     List<Label> labels;
 
     @Schema(additionalProperties = Schema.AdditionalPropertiesValue.TRUE)
@@ -69,4 +75,5 @@ public abstract class AbstractFlow implements FlowInterface {
 
     @Valid
     private WorkerGroup workerGroup;
+
 }

core/src/main/java/io/kestra/core/models/flows/Flow.java

@@ -11,7 +11,6 @@ import com.fasterxml.jackson.databind.introspect.JacksonAnnotationIntrospector;
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.models.HasUID;
 import io.kestra.core.models.annotations.PluginProperty;
-import io.kestra.core.models.flows.lineage.Lineage;
 import io.kestra.core.models.flows.sla.SLA;
 import io.kestra.core.models.listeners.Listener;
 import io.kestra.core.models.tasks.FlowableTask;
@@ -126,9 +125,6 @@ public class Flow extends AbstractFlow implements HasUID {
     @PluginProperty
     List<SLA> sla;
 
-    // TODO add validation
-    Lineage lineage;
-
     public Stream<String> allTypes() {
         return Stream.of(
                 Optional.ofNullable(triggers).orElse(Collections.emptyList()).stream().map(AbstractTrigger::getType),

core/src/main/java/io/kestra/core/models/flows/FlowInterface.java

@@ -10,7 +10,6 @@ import io.kestra.core.models.HasSource;
 import io.kestra.core.models.HasUID;
 import io.kestra.core.models.Label;
 import io.kestra.core.models.TenantInterface;
-import io.kestra.core.models.flows.lineage.Lineage;
 import io.kestra.core.models.flows.sla.SLA;
 import io.kestra.core.models.tasks.WorkerGroup;
 import io.kestra.core.serializers.JacksonMapper;
@@ -56,10 +55,6 @@ public interface FlowInterface extends FlowId, DeletedInterface, TenantInterface
         return List.of();
     }
 
-    default Lineage getLineage() {
-        return null;
-    }
-
     String getSource();
 
     @Override

core/src/main/java/io/kestra/core/models/flows/FlowWithSource.java

@@ -43,7 +43,6 @@ public class FlowWithSource extends Flow {
             .concurrency(this.concurrency)
             .retry(this.retry)
             .sla(this.sla)
-            .lineage(this.lineage)
             .build();
     }
 
@@ -86,7 +85,6 @@ public class FlowWithSource extends Flow {
             .concurrency(flow.concurrency)
             .retry(flow.retry)
             .sla(flow.sla)
-            .lineage(flow.lineage)
             .build();
     }
 }

core/src/main/java/io/kestra/core/models/flows/GenericFlow.java

@@ -9,7 +9,6 @@ import com.google.common.annotations.VisibleForTesting;
 import io.kestra.core.exceptions.DeserializationException;
 import io.kestra.core.models.HasUID;
 import io.kestra.core.models.Label;
-import io.kestra.core.models.flows.lineage.Lineage;
 import io.kestra.core.models.flows.sla.SLA;
 import io.kestra.core.models.tasks.GenericTask;
 import io.kestra.core.models.triggers.GenericTrigger;
@@ -46,8 +45,6 @@ public class GenericFlow extends AbstractFlow implements HasUID {
 
     private Concurrency concurrency;
 
-    Lineage lineage;
-
     private List<GenericTask> tasks;
 
     private List<GenericTrigger> triggers;

core/src/main/java/io/kestra/core/models/flows/State.java

@@ -86,10 +86,11 @@ public class State {
 
     @JsonProperty(access = JsonProperty.Access.READ_ONLY)
     public Duration getDuration() {
-        return Duration.between(
-            this.histories.getFirst().getDate(),
-            this.histories.size() > 1 ? this.histories.get(this.histories.size() - 1).getDate() : Instant.now()
-        );
+        if(this.getEndDate().isPresent()){
+            return Duration.between(this.getStartDate(), this.getEndDate().get());
+        } else {
+            return Duration.between(this.getStartDate(), Instant.now());
+        }
     }
 
     @JsonProperty(access = JsonProperty.Access.READ_ONLY)

core/src/main/java/io/kestra/core/models/flows/lineage/DataSet.java

@@ -1,13 +0,0 @@
-package io.kestra.core.models.flows.lineage;
-
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.experimental.SuperBuilder;
-
-@SuperBuilder
-@Getter
-@NoArgsConstructor
-public class DataSet {
-    private String name;
-    private String namespace;
-}

core/src/main/java/io/kestra/core/models/flows/lineage/Lineage.java

@@ -1,16 +0,0 @@
-package io.kestra.core.models.flows.lineage;
-
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.experimental.SuperBuilder;
-
-import java.util.List;
-
-@SuperBuilder
-@Getter
-@NoArgsConstructor
-public class Lineage {
-    private String namespace;
-    private List<DataSet> inputs;
-    private List<DataSet> outputs;
-}

core/src/main/java/io/kestra/core/models/tasks/ResolvedTask.java

@@ -30,7 +30,7 @@ public class ResolvedTask {
 
     public NextTaskRun toNextTaskRunIncrementIteration(Execution execution, Integer iteration) {
         return new NextTaskRun(
-            TaskRun.of(execution, this).withIteration(iteration != null ? iteration : 1),
+            TaskRun.of(execution, this).withIteration(iteration != null ? iteration : 0),
             this.getTask()
         );
     }

core/src/main/java/io/kestra/core/models/triggers/Trigger.java

@@ -185,34 +185,6 @@ public class Trigger extends TriggerContext implements HasUID {
             .build();
     }
 
-    public static Trigger update(Trigger currentTrigger, Trigger newTrigger, ZonedDateTime nextExecutionDate) throws Exception {
-        Trigger updated = currentTrigger;
-
-        // If a backfill is created, we update the currentTrigger
-        // and set the nextExecutionDate() as the previous one
-        if (newTrigger.getBackfill() != null) {
-            updated = currentTrigger.toBuilder()
-                .backfill(
-                    newTrigger
-                        .getBackfill()
-                        .toBuilder()
-                        .end(newTrigger.getBackfill().getEnd() != null ? newTrigger.getBackfill().getEnd() : ZonedDateTime.now())
-                        .currentDate(
-                            newTrigger.getBackfill().getStart()
-                        )
-                        .previousNextExecutionDate(
-                            currentTrigger.getNextExecutionDate())
-                        .build())
-                .build();
-        }
-
-        return updated.toBuilder()
-            .nextExecutionDate(newTrigger.getDisabled() ?
-                null : nextExecutionDate)
-            .disabled(newTrigger.getDisabled())
-            .build();
-    }
-
     public Trigger resetExecution(Flow flow, Execution execution, ConditionContext conditionContext) {
         boolean disabled = this.getStopAfter() != null ? this.getStopAfter().contains(execution.getState().getCurrent()) : this.getDisabled();
         if (!disabled) {
@@ -276,27 +248,22 @@ public class Trigger extends TriggerContext implements HasUID {
             .build();
     }
 
-    public Trigger initBackfill(Trigger newTrigger) {
-        // If a backfill is created, we update the currentTrigger
+    public Trigger withBackfill(final Backfill backfill) {
+        Trigger updated = this;
+        // If a backfill is created, we update the trigger
         // and set the nextExecutionDate() as the previous one
-        if (newTrigger.getBackfill() != null) {
-
-            return this.toBuilder()
+        if (backfill != null) {
+            updated = this.toBuilder()
                 .backfill(
-                    newTrigger
-                        .getBackfill()
+                    backfill
                         .toBuilder()
-                        .end(newTrigger.getBackfill().getEnd() != null ? newTrigger.getBackfill().getEnd() : ZonedDateTime.now())
-                        .currentDate(
-                            newTrigger.getBackfill().getStart()
-                        )
-                        .previousNextExecutionDate(
-                            this.getNextExecutionDate())
+                        .end(backfill.getEnd() != null ? backfill.getEnd() : ZonedDateTime.now())
+                        .currentDate(backfill.getStart())
+                        .previousNextExecutionDate(this.getNextExecutionDate())
                         .build())
                 .build();
         }
-
-        return this;
+        return updated;
     }
 
     // if the next date is after the backfill end, we remove the backfill

core/src/main/java/io/kestra/core/queues/QueueFactoryInterface.java

@@ -27,6 +27,7 @@ public interface QueueFactoryInterface {
     String CLUSTER_EVENT_NAMED = "clusterEventQueue";
     String SUBFLOWEXECUTIONEND_NAMED = "subflowExecutionEndQueue";
     String EXECUTION_RUNNING_NAMED = "executionRunningQueue";
+    String MULTIPLE_CONDITION_EVENT_NAMED = "multipleConditionEventQueue";
 
     QueueInterface<Execution> execution();
 
@@ -59,4 +60,6 @@ public interface QueueFactoryInterface {
     QueueInterface<SubflowExecutionEnd> subflowExecutionEnd();
 
     QueueInterface<ExecutionRunning> executionRunning();
+
+    QueueInterface<MultipleConditionEvent> multipleConditionEvent();
 }

core/src/main/java/io/kestra/core/repositories/ExecutionRepositoryInterface.java

@@ -25,8 +25,6 @@ import java.util.Optional;
 import java.util.function.Function;
 
 public interface ExecutionRepositoryInterface extends SaveRepositoryInterface<Execution>, QueryBuilderInterface<Executions.Fields> {
-    Boolean isTaskRunEnabled();
-
     default Optional<Execution> findById(String tenantId, String id) {
         return findById(tenantId, id, false);
     }
@@ -96,24 +94,19 @@ public interface ExecutionRepositoryInterface extends SaveRepositoryInterface<Ex
 
     Flux<Execution> findAllAsync(@Nullable String tenantId);
 
-    ArrayListTotal<TaskRun> findTaskRun(
-        Pageable pageable,
-        @Nullable String tenantId,
-        List<QueryFilter> filters
-    );
-
     Execution delete(Execution execution);
 
     Integer purge(Execution execution);
 
+    Integer purge(List<Execution> executions);
+
     List<DailyExecutionStatistics> dailyStatisticsForAllTenants(
         @Nullable String query,
         @Nullable String namespace,
         @Nullable String flowId,
         @Nullable ZonedDateTime startDate,
         @Nullable ZonedDateTime endDate,
-        @Nullable DateUtils.GroupType groupBy,
-        boolean isTaskRun
+        @Nullable DateUtils.GroupType groupBy
     );
 
     List<DailyExecutionStatistics> dailyStatistics(
@@ -125,8 +118,7 @@ public interface ExecutionRepositoryInterface extends SaveRepositoryInterface<Ex
         @Nullable ZonedDateTime startDate,
         @Nullable ZonedDateTime endDate,
         @Nullable DateUtils.GroupType groupBy,
-        List<State.Type> state,
-        boolean isTaskRun
+        List<State.Type> state
     );
 
     @Getter

core/src/main/java/io/kestra/core/repositories/FlowRepositoryInterface.java

@@ -4,6 +4,7 @@ import io.kestra.core.models.QueryFilter;
 import io.kestra.core.models.SearchResult;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.*;
+import io.kestra.plugin.core.dashboard.data.Flows;
 import io.micronaut.data.model.Pageable;
 import jakarta.annotation.Nullable;
 import jakarta.validation.ConstraintViolationException;
@@ -11,7 +12,7 @@ import jakarta.validation.ConstraintViolationException;
 import java.util.List;
 import java.util.Optional;
 
-public interface FlowRepositoryInterface {
+public interface FlowRepositoryInterface extends QueryBuilderInterface<Flows.Fields> {
 
     Optional<Flow> findById(String tenantId, String namespace, String id, Optional<Integer> revision, Boolean allowDeleted);
 
@@ -162,4 +163,6 @@ public interface FlowRepositoryInterface {
     FlowWithSource update(GenericFlow flow, FlowInterface previous) throws ConstraintViolationException;
 
     FlowWithSource delete(FlowInterface flow);
+
+    Boolean existAnyNoAcl(String tenantId);
 }

core/src/main/java/io/kestra/core/repositories/LogRepositoryInterface.java

@@ -90,6 +90,8 @@ public interface LogRepositoryInterface extends SaveRepositoryInterface<LogEntry
 
     Integer purge(Execution execution);
 
+    Integer purge(List<Execution> executions);
+
     void deleteByQuery(String tenantId, String executionId, String taskId, String taskRunId, Level minLevel, Integer attempt);
 
     void deleteByQuery(String tenantId, String namespace, String flowId, String triggerId);

core/src/main/java/io/kestra/core/repositories/MetricRepositoryInterface.java

@@ -29,6 +29,8 @@ public interface MetricRepositoryInterface extends SaveRepositoryInterface<Metri
 
     Integer purge(Execution execution);
 
+    Integer purge(List<Execution> executions);
+
     Flux<MetricEntry> findAllAsync(@Nullable String tenantId);
 
     default Function<String, String> sortMapping() throws IllegalArgumentException {

core/src/main/java/io/kestra/core/runners/ExecutableUtils.java

@@ -5,10 +5,7 @@ import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.models.Label;
 import io.kestra.core.models.executions.*;
-import io.kestra.core.models.flows.Flow;
-import io.kestra.core.models.flows.FlowInterface;
-import io.kestra.core.models.flows.FlowWithException;
-import io.kestra.core.models.flows.State;
+import io.kestra.core.models.flows.*;
 import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.ExecutableTask;
 import io.kestra.core.models.tasks.Task;
@@ -29,6 +26,7 @@ import org.apache.commons.lang3.stream.Streams;
 import java.time.Instant;
 import java.time.ZonedDateTime;
 import java.util.*;
+import java.util.stream.Collectors;
 
 import static io.kestra.core.trace.Tracer.throwCallable;
 import static io.kestra.core.utils.Rethrow.throwConsumer;
@@ -153,14 +151,22 @@ public final class ExecutableUtils {
                     currentFlow.getNamespace(),
                     currentFlow.getId()
                 )
-                .orElseThrow(() -> new IllegalStateException("Unable to find flow '" + subflowNamespace + "'.'" + subflowId + "' with revision '" + subflowRevision.orElse(0) + "'"));
+                .orElseThrow(() -> {
+                    String msg = "Unable to find flow '" + subflowNamespace + "'.'" + subflowId + "' with revision '" + subflowRevision.orElse(0) + "'";
+                    runContext.logger().error(msg);
+                    return new IllegalStateException(msg);
+                });
 
             if (flow.isDisabled()) {
-                throw new IllegalStateException("Cannot execute a flow which is disabled");
+                String msg = "Cannot execute a flow which is disabled";
+                runContext.logger().error(msg);
+                throw new IllegalStateException(msg);
             }
 
             if (flow instanceof FlowWithException fwe) {
-                throw new IllegalStateException("Cannot execute an invalid flow: " + fwe.getException());
+                String msg = "Cannot execute an invalid flow: " + fwe.getException();
+                runContext.logger().error(msg);
+                throw new IllegalStateException(msg);
             }
 
             List<Label> newLabels = inheritLabels ? new ArrayList<>(filterLabels(currentExecution.getLabels(), flow)) : new ArrayList<>(systemLabels(currentExecution));
@@ -201,7 +207,20 @@ public final class ExecutableUtils {
                     .build()
                 )
                 .withScheduleDate(scheduleOnDate);
-
+                if(execution.getInputs().size()<inputs.size()) {
+                    Map<String,Object>resolvedInputs=execution.getInputs();
+                    for (var inputKey : inputs.keySet()) {
+                        if (!resolvedInputs.containsKey(inputKey)) {
+                            runContext.logger().warn(
+                                "Input {} was provided by parent execution {} for subflow {}.{} but isn't declared at the subflow inputs",
+                                inputKey,
+                                currentExecution.getId(),
+                                currentTask.subflowId().namespace(),
+                                currentTask.subflowId().flowId()
+                            );
+                        }
+                    }
+                }
             // inject the traceparent into the new execution
             propagator.ifPresent(pg -> pg.inject(Context.current(), execution, ExecutionTextMapSetter.INSTANCE));
 

core/src/main/java/io/kestra/core/runners/ExecutionRunning.java

@@ -32,5 +32,7 @@ public class ExecutionRunning implements HasUID {
         return IdUtils.fromPartsAndSeparator('|', this.tenantId, this.namespace, this.flowId, this.execution.getId());
     }
 
-    public enum ConcurrencyState { CREATED, RUNNING, QUEUED, CANCELLED, FAILED }
+    // Note: the KILLED state is only used in the Kafka implementation to difference between purging terminated running execution (null)
+    // and purging killed execution which need special treatment
+    public enum ConcurrencyState { CREATED, RUNNING, QUEUED, CANCELLED, FAILED, KILLED }
 }

core/src/main/java/io/kestra/core/runners/FlowInputOutput.java

@@ -49,15 +49,7 @@ import java.time.Duration;
 import java.time.Instant;
 import java.time.LocalDate;
 import java.time.LocalTime;
-import java.util.AbstractMap;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
+import java.util.*;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.regex.Matcher;
@@ -231,6 +223,19 @@ public class FlowInputOutput {
                 return new AbstractMap.SimpleEntry<>(it.input().getId(), it.value());
             })
             .collect(HashMap::new, (m,v)-> m.put(v.getKey(), v.getValue()), HashMap::putAll);
+        if (resolved.size() < data.size()) {
+            RunContext runContext = runContextFactory.of(flow, execution);
+            for (var inputKey : data.keySet()) {
+                if (!resolved.containsKey(inputKey)) {
+                    runContext.logger().warn(
+                        "Input {} was provided for workflow {}.{} but isn't declared in the workflow inputs",
+                        inputKey,
+                        flow.getNamespace(),
+                        flow.getId()
+                    );
+                }
+            }
+        }
         return MapUtils.flattenToNestedMap(resolved);
     }
 
@@ -313,15 +318,15 @@ public class FlowInputOutput {
             });
             resolvable.setInput(input);
 
-            
+
             Object value = resolvable.get().value();
-            
+
             // resolve default if needed
             if (value == null && input.getDefaults() != null) {
                 value = resolveDefaultValue(input, runContext);
                 resolvable.isDefault(true);
             }
-            
+
             // validate and parse input value
             if (value == null) {
                 if (input.getRequired()) {
@@ -350,7 +355,7 @@ public class FlowInputOutput {
 
         return resolvable.get();
     }
-    
+
     public static Object resolveDefaultValue(Input<?> input, PropertyContext renderer) throws IllegalVariableEvaluationException {
         return switch (input.getType()) {
             case STRING, ENUM, SELECT, SECRET, EMAIL -> resolveDefaultPropertyAs(input, renderer, String.class);
@@ -367,7 +372,7 @@ public class FlowInputOutput {
             case MULTISELECT -> resolveDefaultPropertyAsList(input, renderer, String.class);
         };
     }
-    
+
     @SuppressWarnings("unchecked")
     private static <T> Object resolveDefaultPropertyAs(Input<?> input, PropertyContext renderer, Class<T> clazz) throws IllegalVariableEvaluationException {
         return Property.as((Property<T>) input.getDefaults(), renderer, clazz);
@@ -376,7 +381,7 @@ public class FlowInputOutput {
     private static <T> Object resolveDefaultPropertyAsList(Input<?> input, PropertyContext renderer, Class<T> clazz) throws IllegalVariableEvaluationException {
         return Property.asList((Property<List<T>>) input.getDefaults(), renderer, clazz);
     }
-    
+
     private RunContext buildRunContextForExecutionAndInputs(final FlowInterface flow, final Execution execution, Map<String, InputAndValue> dependencies) {
         Map<String, Object> flattenInputs = MapUtils.flattenToNestedMap(dependencies.entrySet()
             .stream()
@@ -453,7 +458,7 @@ public class FlowInputOutput {
         if (data.getType() == null) {
             return Optional.of(new AbstractMap.SimpleEntry<>(data.getId(), current));
         }
-        
+
         final Type elementType = data instanceof ItemTypeInterface itemTypeInterface ? itemTypeInterface.getItemType() : null;
 
         return Optional.of(new AbstractMap.SimpleEntry<>(
@@ -530,17 +535,17 @@ public class FlowInputOutput {
             throw new Exception("Expected `" + type + "` but received `" + current + "` with errors:\n```\n" + e.getMessage() + "\n```");
         }
     }
-    
+
     public static Map<String, Object> renderFlowOutputs(List<Output> outputs, RunContext runContext) throws IllegalVariableEvaluationException {
         if (outputs == null) return Map.of();
-        
+
         // render required outputs
         Map<String, Object> outputsById = outputs
             .stream()
             .filter(output -> output.getRequired() == null || output.getRequired())
             .collect(HashMap::new, (map, entry) -> map.put(entry.getId(), entry.getValue()), Map::putAll);
         outputsById = runContext.render(outputsById);
-        
+
         // render optional outputs one by one to catch, log, and skip any error.
         for (io.kestra.core.models.flows.Output output : outputs) {
             if (Boolean.FALSE.equals(output.getRequired())) {
@@ -583,9 +588,9 @@ public class FlowInputOutput {
         }
 
         public void isDefault(boolean isDefault) {
-            this.input = new InputAndValue(this.input.input(), this.input.value(), this.input.enabled(), isDefault, this.input.exception());  
+            this.input = new InputAndValue(this.input.input(), this.input.value(), this.input.enabled(), isDefault, this.input.exception());
         }
-        
+
         public void setInput(final Input<?> input) {
             this.input = new InputAndValue(input, this.input.value(), this.input.enabled(), this.input.isDefault(), this.input.exception());
         }

core/src/main/java/io/kestra/core/runners/FlowableUtils.java

@@ -500,7 +500,7 @@ public class FlowableUtils {
 
         ArrayList<ResolvedTask> result = new ArrayList<>();
 
-        int index = 0;
+        int iteration = 0;
         for (Object current : distinctValue) {
             try {
                 String resolvedValue = current instanceof String stringValue ? stringValue : MAPPER.writeValueAsString(current);
@@ -508,7 +508,7 @@ public class FlowableUtils {
                     result.add(ResolvedTask.builder()
                         .task(task)
                         .value(resolvedValue)
-                        .iteration(index++)
+                        .iteration(iteration)
                         .parentId(parentTaskRun.getId())
                         .build()
                     );
@@ -516,6 +516,7 @@ public class FlowableUtils {
             } catch (JsonProcessingException e) {
                 throw new IllegalVariableEvaluationException(e);
             }
+            iteration++;
         }
 
         return result;

core/src/main/java/io/kestra/core/runners/MultipleConditionEvent.java

@@ -0,0 +1,13 @@
+package io.kestra.core.runners;
+
+import io.kestra.core.models.HasUID;
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.utils.IdUtils;
+
+public record MultipleConditionEvent(Flow flow, Execution execution) implements HasUID {
+    @Override
+    public String uid() {
+        return IdUtils.fromParts(flow.uidWithoutRevision(), execution.getId());
+    }
+}

core/src/main/java/io/kestra/core/runners/pebble/filters/TimestampMicroFilter.java

@@ -30,6 +30,6 @@ public class TimestampMicroFilter extends AbstractDate implements Filter {
         ZoneId zoneId = zoneId(timeZone);
         ZonedDateTime date = convert(input, zoneId, existingFormat);
 
-        return String.valueOf(TimeUnit.SECONDS.toNanos(date.toEpochSecond()) + TimeUnit.NANOSECONDS.toMicros(date.getNano()));
+        return String.valueOf(TimeUnit.SECONDS.toMicros(date.toEpochSecond()) + TimeUnit.NANOSECONDS.toMicros(date.getNano()));
     }
 }

core/src/main/java/io/kestra/core/services/ExecutionService.java

@@ -56,8 +56,7 @@ import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
-import static io.kestra.core.utils.Rethrow.throwFunction;
-import static io.kestra.core.utils.Rethrow.throwPredicate;
+import static io.kestra.core.utils.Rethrow.*;
 
 @Singleton
 @Slf4j
@@ -431,7 +430,8 @@ public class ExecutionService {
         @Nullable String flowId,
         @Nullable ZonedDateTime startDate,
         @Nullable ZonedDateTime endDate,
-        @Nullable List<State.Type> state
+        @Nullable List<State.Type> state,
+        int batchSize
     ) throws IOException {
         PurgeResult purgeResult = this.executionRepository
             .find(
@@ -448,24 +448,27 @@ public class ExecutionService {
                 null,
                 true
             )
-            .map(throwFunction(execution -> {
+            .buffer(batchSize)
+            .map(throwFunction(executions -> {
                 PurgeResult.PurgeResultBuilder<?, ?> builder = PurgeResult.builder();
 
                 if (purgeExecution) {
-                    builder.executionsCount(this.executionRepository.purge(execution));
+                    builder.executionsCount(this.executionRepository.purge(executions));
                 }
 
                 if (purgeLog) {
-                    builder.logsCount(this.logRepository.purge(execution));
+                    builder.logsCount(this.logRepository.purge(executions));
                 }
 
                 if (purgeMetric) {
-                    builder.metricsCount(this.metricRepository.purge(execution));
+                    builder.metricsCount(this.metricRepository.purge(executions));
                 }
 
                 if (purgeStorage) {
-                    URI uri = StorageContext.forExecution(execution).getExecutionStorageURI(StorageContext.KESTRA_SCHEME);
-                    builder.storagesCount(storageInterface.deleteByPrefix(execution.getTenantId(), execution.getNamespace(), uri).size());
+                    executions.forEach(throwConsumer(execution -> {
+                        URI uri = StorageContext.forExecution(execution).getExecutionStorageURI(StorageContext.KESTRA_SCHEME);
+                        builder.storagesCount(storageInterface.deleteByPrefix(execution.getTenantId(), execution.getNamespace(), uri).size());
+                    }));
                 }
 
                 return (PurgeResult) builder.build();
@@ -716,7 +719,8 @@ public class ExecutionService {
             newExecution = execution.withState(killingOrAfterKillState);
         }
 
-        eventPublisher.publishEvent(new CrudEvent<>(newExecution, execution, CrudEventType.UPDATE));
+        // Because this method is expected to be called by the Executor we can return the Execution
+        // immediately without publishing a CrudEvent like it's done on pause/resume method.
         return newExecution;
     }
     public Execution kill(Execution execution, FlowInterface flow) {

core/src/main/java/io/kestra/core/services/SkipExecutionService.java

@@ -1,4 +1,4 @@
-package io.kestra.executor;
+package io.kestra.core.services;
 
 import com.google.common.annotations.VisibleForTesting;
 import io.kestra.core.models.executions.Execution;
@@ -18,6 +18,7 @@ public class SkipExecutionService {
     private volatile List<FlowId> skipFlows = Collections.emptyList();
     private volatile List<NamespaceId> skipNamespaces = Collections.emptyList();
     private volatile List<String> skipTenants = Collections.emptyList();
+    private volatile List<String> skipIndexerRecords = Collections.emptyList();
 
     public synchronized void setSkipExecutions(List<String> skipExecutions) {
         this.skipExecutions = skipExecutions == null ? Collections.emptyList() : skipExecutions;
@@ -35,6 +36,10 @@ public class SkipExecutionService {
         this.skipTenants = skipTenants == null ? Collections.emptyList() : skipTenants;
     }
 
+    public synchronized void setSkipIndexerRecords(List<String> skipIndexerRecords) {
+        this.skipIndexerRecords = skipIndexerRecords == null ? Collections.emptyList() : skipIndexerRecords;
+    }
+
     /**
      * Warning: this method didn't check the flow, so it must be used only when neither of the others can be used.
      */
@@ -50,6 +55,14 @@ public class SkipExecutionService {
         return skipExecution(taskRun.getTenantId(), taskRun.getNamespace(), taskRun.getFlowId(), taskRun.getExecutionId());
     }
 
+    /**
+     * Skip an indexer records based on its key.
+     * @param key the record key as computed by <code>QueueService.key(record)</code>, can be null
+     */
+    public boolean skipIndexerRecord(@Nullable String key) {
+        return key != null && skipIndexerRecords.contains(key);
+    }
+
     @VisibleForTesting
     boolean skipExecution(String tenant, String namespace, String flow, String executionId) {
         return (tenant != null && skipTenants.contains(tenant)) ||

core/src/main/java/io/kestra/core/utils/VersionProvider.java

@@ -46,16 +46,19 @@ public class VersionProvider {
         this.date = loadTime(gitProperties);
         this.version = loadVersion(buildProperties, gitProperties);
 
-        // check the version in the settings and update if needed, we did't use it would allow us to detect incompatible update later if needed
-        if (settingRepository.isPresent()) {
-            Optional<Setting> versionSetting = settingRepository.get().findByKey(Setting.INSTANCE_VERSION);
-            if (versionSetting.isEmpty() || !versionSetting.get().getValue().equals(this.version)) {
-                settingRepository.get().save(Setting.builder()
-                    .key(Setting.INSTANCE_VERSION)
-                    .value(this.version)
-                    .build()
-                );
-            }
+        // check the version in the settings and update if needed, we didn't use it would allow us to detect incompatible update later if needed
+        settingRepository.ifPresent(
+            settingRepositoryInterface -> persistVersion(settingRepositoryInterface, version));
+    }
+
+    private static synchronized void persistVersion(SettingRepositoryInterface settingRepositoryInterface, String version) {
+        Optional<Setting> versionSetting = settingRepositoryInterface.findByKey(Setting.INSTANCE_VERSION);
+        if (versionSetting.isEmpty() || !versionSetting.get().getValue().equals(version)) {
+            settingRepositoryInterface.save(Setting.builder()
+                .key(Setting.INSTANCE_VERSION)
+                .value(version)
+                .build()
+            );
         }
     }
 

core/src/main/java/io/kestra/core/validations/validator/FlowValidator.java

@@ -63,7 +63,7 @@ public class FlowValidator implements ConstraintValidator<FlowValidation, Flow>
 
         List<String> violations = new ArrayList<>();
 
-        if (RESERVED_FLOW_IDS.contains(value.getId())) {
+        if (value.getId() != null && RESERVED_FLOW_IDS.contains(value.getId())) {
             violations.add("Flow id is a reserved keyword: " + value.getId() + ". List of reserved keywords: " + String.join(", ", RESERVED_FLOW_IDS));
         }
 

core/src/main/java/io/kestra/plugin/core/dashboard/data/Flows.java

@@ -0,0 +1,52 @@
+package io.kestra.plugin.core.dashboard.data;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import io.kestra.core.models.annotations.Example;
+import io.kestra.core.models.annotations.Plugin;
+import io.kestra.core.models.dashboards.ColumnDescriptor;
+import io.kestra.core.models.dashboards.DataFilter;
+import io.kestra.core.repositories.FlowRepositoryInterface;
+import io.kestra.core.repositories.QueryBuilderInterface;
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+@SuperBuilder(toBuilder = true)
+@Getter
+@NoArgsConstructor
+@JsonInclude(JsonInclude.Include.NON_DEFAULT)
+@EqualsAndHashCode
+@Schema(
+    title = "Display Flow data in a dashboard chart."
+)
+@Plugin(
+    examples = {
+        @Example(
+            title = "Display a chart with a list of Flows.",
+            full = true,
+            code = { """
+            charts:
+              - id: list_flows
+                type: io.kestra.plugin.core.dashboard.chart.Table
+                data:
+                  type: io.kestra.plugin.core.dashboard.data.Flows
+                  columns:
+                    namespace:
+                      field: NAMESPACE
+                    id:
+                      field: ID
+            """
+          }
+        )
+    }
+)
+@JsonTypeName("Flows")
+public class Flows<C extends ColumnDescriptor<Flows.Fields>> extends DataFilter<Flows.Fields, C> implements IFlows {
+    @Override
+    public Class<? extends QueryBuilderInterface<Fields>> repositoryClass() {
+        return FlowRepositoryInterface.class;
+    }
+}

core/src/main/java/io/kestra/plugin/core/dashboard/data/FlowsKPI.java

@@ -0,0 +1,51 @@
+package io.kestra.plugin.core.dashboard.data;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import io.kestra.core.models.annotations.Example;
+import io.kestra.core.models.annotations.Plugin;
+import io.kestra.core.models.dashboards.ColumnDescriptor;
+import io.kestra.core.models.dashboards.DataFilterKPI;
+import io.kestra.core.repositories.FlowRepositoryInterface;
+import io.kestra.core.repositories.QueryBuilderInterface;
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+@SuperBuilder(toBuilder = true)
+@Getter
+@NoArgsConstructor
+@JsonInclude(JsonInclude.Include.NON_DEFAULT)
+@EqualsAndHashCode
+@Schema(
+    title = "Display a chart with Flows KPI.",
+    description = "Change."
+)
+@Plugin(
+    examples = {
+        @Example(
+            title = "Display count of Flows.",
+            full = true,
+            code = { """
+            charts:
+              - id: kpi
+                    type: io.kestra.plugin.core.dashboard.chart.KPI
+                    data:
+                      type: io.kestra.plugin.core.dashboard.data.FlowsKPI
+                      columns:
+                        field: ID
+                        agg: COUNT
+            """
+          }
+        )
+    }
+)
+@JsonTypeName("FlowsKPI")
+public class FlowsKPI<C extends ColumnDescriptor<FlowsKPI.Fields>> extends DataFilterKPI<FlowsKPI.Fields, C> implements IFlows {
+    @Override
+    public Class<? extends QueryBuilderInterface<Fields>> repositoryClass() {
+        return FlowRepositoryInterface.class;
+    }
+}

core/src/main/java/io/kestra/plugin/core/dashboard/data/IFlows.java

@@ -0,0 +1,37 @@
+package io.kestra.plugin.core.dashboard.data;
+
+import io.kestra.core.models.QueryFilter;
+import io.kestra.core.models.dashboards.filters.AbstractFilter;
+import io.kestra.core.utils.ListUtils;
+
+import java.time.ZonedDateTime;
+import java.util.ArrayList;
+import java.util.List;
+
+public interface IFlows extends IData<IFlows.Fields> {
+
+    default List<AbstractFilter<IFlows.Fields>> whereWithGlobalFilters(List<QueryFilter> filters, ZonedDateTime startDate, ZonedDateTime endDate, List<AbstractFilter<IFlows.Fields>> where) {
+        List<AbstractFilter<IFlows.Fields>> updatedWhere = where != null ? new ArrayList<>(where) : new ArrayList<>();
+
+        if (ListUtils.isEmpty(filters)) {
+            return updatedWhere;
+        }
+
+        List<QueryFilter> namespaceFilters = filters.stream().filter(f -> f.field().equals(QueryFilter.Field.NAMESPACE)).toList();
+        if (!namespaceFilters.isEmpty()) {
+            updatedWhere.removeIf(filter -> filter.getField().equals(IFlows.Fields.NAMESPACE));
+            namespaceFilters.forEach(f -> {
+                updatedWhere.add(f.toDashboardFilterBuilder(IFlows.Fields.NAMESPACE, f.value()));
+            });
+        }
+
+
+        return updatedWhere;
+    }
+
+    enum Fields {
+        ID,
+        NAMESPACE,
+        REVISION
+    }
+}

core/src/main/java/io/kestra/plugin/core/execution/Labels.java

@@ -155,6 +155,7 @@ public class Labels extends Task implements ExecutionUpdatableTask {
         newLabels.putAll(labelsAsMap);
 
         return execution.withLabels(newLabels.entrySet().stream()
+            .filter(Label.getEntryNotEmptyPredicate())
             .map(entry -> new Label(
                 entry.getKey(),
                 entry.getValue()

core/src/main/java/io/kestra/plugin/core/execution/PurgeExecutions.java

@@ -102,6 +102,14 @@ public class PurgeExecutions extends Task implements RunnableTask<PurgeExecution
     @Builder.Default
     private Property<Boolean> purgeStorage = Property.ofValue(true);
 
+    @Schema(
+        title = "The size of the bulk delete",
+        description = "For performance, deletion is made by batch of by default 100 executions/logs/metrics."
+    )
+    @Builder.Default
+    @NotNull
+    private Property<Integer> batchSize = Property.ofValue(100);
+
     @Override
     public PurgeExecutions.Output run(RunContext runContext) throws Exception {
         ExecutionService executionService = ((DefaultRunContext)runContext).getApplicationContext().getBean(ExecutionService.class);
@@ -124,9 +132,10 @@ public class PurgeExecutions extends Task implements RunnableTask<PurgeExecution
             flowInfo.tenantId(),
             renderedNamespace,
             runContext.render(flowId).as(String.class).orElse(null),
-            startDate != null ? ZonedDateTime.parse(runContext.render(startDate).as(String.class).orElseThrow()) : null,
+            runContext.render(startDate).as(String.class).map(ZonedDateTime::parse).orElse(null),
             ZonedDateTime.parse(runContext.render(endDate).as(String.class).orElseThrow()),
-            this.states == null ? null : runContext.render(this.states).asList(State.Type.class)
+            this.states == null ? null : runContext.render(this.states).asList(State.Type.class),
+            runContext.render(this.batchSize).as(Integer.class).orElseThrow()
         );
 
         return Output.builder()

core/src/main/java/io/kestra/plugin/core/flow/ForEachItem.java

@@ -478,7 +478,7 @@ public class ForEachItem extends Task implements FlowableTask<VoidOutput>, Child
             try (InputStream is = runContext.storage().getFile(splitsURI)){
                 String fileContent = new String(is.readAllBytes());
                 List<URI> splits = fileContent.lines().map(line -> URI.create(line)).toList();
-                AtomicInteger currentIteration = new AtomicInteger(1);
+                AtomicInteger currentIteration = new AtomicInteger(0);
 
                 return splits
                     .stream()

core/src/test/java/io/kestra/core/models/LabelTest.java

@@ -1,19 +1,32 @@
 package io.kestra.core.models;
 
+import io.kestra.core.junit.annotations.KestraTest;
+import io.kestra.core.models.validations.ModelValidator;
+import jakarta.inject.Inject;
+import jakarta.validation.ConstraintViolationException;
 import org.junit.jupiter.api.Test;
 
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+@KestraTest
 class LabelTest {
+    @Inject
+    private ModelValidator modelValidator;
 
     @Test
     void shouldGetNestedMapGivenDistinctLabels() {
         Map<String, Object> result = Label.toNestedMap(List.of(
             new Label(Label.USERNAME, "test"),
-            new Label(Label.CORRELATION_ID, "id"))
+            new Label(Label.CORRELATION_ID, "id"),
+            new Label("", "bar"),
+            new Label(null, "bar"),
+            new Label("foo", ""),
+            new Label("baz", null)
+            )
         );
 
         assertThat(result).isEqualTo(
@@ -34,6 +47,18 @@ class LabelTest {
         );
     }
 
+    @Test
+    void toNestedMapShouldIgnoreEmptyOrNull() {
+        Map<String, Object> result = Label.toNestedMap(List.of(
+            new Label("", "bar"),
+            new Label(null, "bar"),
+            new Label("foo", ""),
+            new Label("baz", null))
+        );
+
+        assertThat(result).isEmpty();
+    }
+
     @Test
     void shouldGetMapGivenDistinctLabels() {
         Map<String, String> result = Label.toMap(List.of(
@@ -59,6 +84,18 @@ class LabelTest {
         );
     }
 
+    @Test
+    void toMapShouldIgnoreEmptyOrNull() {
+        Map<String, String> result = Label.toMap(List.of(
+            new Label("", "bar"),
+            new Label(null, "bar"),
+            new Label("foo", ""),
+            new Label("baz", null))
+        );
+
+        assertThat(result).isEmpty();
+    }
+
     @Test
     void shouldDuplicateLabelsWithKeyOrderKept() {
         List<Label> result = Label.deduplicate(List.of(
@@ -73,4 +110,28 @@ class LabelTest {
             new Label(Label.CORRELATION_ID, "id")
         );
     }
+
+    @Test
+    void deduplicateShouldIgnoreEmptyAndNull() {
+        List<Label> result = Label.deduplicate(List.of(
+            new Label("", "bar"),
+            new Label(null, "bar"),
+            new Label("foo", ""),
+            new Label("baz", null))
+        );
+
+        assertThat(result).isEmpty();
+    }
+
+    @Test
+    void shouldValidateEmpty() {
+        Optional<ConstraintViolationException> validLabelResult = modelValidator.isValid(new Label("foo", "bar"));
+        assertThat(validLabelResult.isPresent()).isFalse();
+
+        Optional<ConstraintViolationException> emptyValueLabelResult = modelValidator.isValid(new Label("foo", ""));
+        assertThat(emptyValueLabelResult.isPresent()).isTrue();
+
+        Optional<ConstraintViolationException> emptyKeyLabelResult = modelValidator.isValid(new Label("", "bar"));
+        assertThat(emptyKeyLabelResult.isPresent()).isTrue();
+    }
 }

core/src/test/java/io/kestra/core/models/executions/StateDurationTest.java

@@ -0,0 +1,55 @@
+package io.kestra.core.models.executions;
+
+import io.kestra.core.models.flows.State;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class StateDurationTest {
+
+
+    private static final Instant NOW = Instant.now();
+    private static final Instant ONE = NOW.minus(Duration.ofDays(1000));
+    private static final Instant TWO = ONE.plus(Duration.ofHours(11));
+    private static final Instant THREE = TWO.plus(Duration.ofHours(222));
+
+    @Test
+    void justCreated() {
+        var state = State.of(
+            State.Type.CREATED,
+            List.of(
+                new State.History(State.Type.CREATED, ONE)
+            )
+        );
+        assertThat(state.getDuration()).isCloseTo(Duration.between(ONE, NOW), Duration.ofMinutes(10));
+    }
+
+    @Test
+    void success() {
+        var state = State.of(
+            State.Type.SUCCESS,
+            List.of(
+                new State.History(State.Type.CREATED, ONE),
+                new State.History(State.Type.RUNNING, TWO),
+                new State.History(State.Type.SUCCESS, THREE)
+            )
+        );
+        assertThat(state.getDuration()).isEqualTo(Duration.between(ONE, THREE));
+    }
+
+    @Test
+    void isRunning() {
+        var state = State.of(
+            State.Type.RUNNING,
+            List.of(
+                new State.History(State.Type.CREATED, ONE),
+                new State.History(State.Type.RUNNING, TWO)
+            )
+        );
+        assertThat(state.getDuration()).isCloseTo(Duration.between(ONE, NOW), Duration.ofMinutes(10));
+    }
+}

core/src/test/java/io/kestra/core/models/hierarchies/FlowGraphTest.java

@@ -7,12 +7,11 @@ import io.kestra.core.junit.annotations.ExecuteFlow;
 import io.kestra.core.junit.annotations.KestraTest;
 import io.kestra.core.junit.annotations.LoadFlows;
 import io.kestra.core.models.executions.Execution;
-import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.FlowWithSource;
 import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.queues.QueueException;
 import io.kestra.core.repositories.TriggerRepositoryInterface;
-import io.kestra.core.runners.RunnerUtils;
+import io.kestra.core.runners.TestRunnerUtils;
 import io.kestra.core.serializers.YamlParser;
 import io.kestra.core.services.GraphService;
 import io.kestra.core.utils.GraphUtils;
@@ -45,7 +44,7 @@ class FlowGraphTest {
     private TriggerRepositoryInterface triggerRepositoryInterface;
 
     @Inject
-    private RunnerUtils runnerUtils;
+    private TestRunnerUtils runnerUtils;
 
     @Test
     void simple() throws IllegalVariableEvaluationException, IOException {

core/src/test/java/io/kestra/core/reporter/reports/AbstractFeatureUsageReportTest.java

@@ -13,21 +13,20 @@ import static org.assertj.core.api.Assertions.assertThat;
 
 @KestraTest
 public abstract class AbstractFeatureUsageReportTest {
-    
+
     @Inject
     FeatureUsageReport featureUsageReport;
-    
+
     @Test
     public void shouldGetReport() {
         // When
         Instant now = Instant.now();
         FeatureUsageReport.UsageEvent event = featureUsageReport.report(
-            now, 
+            now,
             Reportable.TimeInterval.of(now.minus(Duration.ofDays(1)).atZone(ZoneId.systemDefault()), now.atZone(ZoneId.systemDefault()))
         );
-        
+
         // Then
         assertThat(event.getExecutions().getDailyExecutionsCount().size()).isGreaterThan(0);
-        assertThat(event.getExecutions().getDailyTaskRunsCount()).isNull();
     }
 }

core/src/test/java/io/kestra/core/reporter/reports/AbstractServiceUsageReportTest.java

@@ -22,24 +22,24 @@ import java.util.Set;
 
 @KestraTest
 public abstract class AbstractServiceUsageReportTest {
-    
+
     @Inject
     ServiceUsageReport serviceUsageReport;
-    
+
     @Inject
     ServiceInstanceRepositoryInterface serviceInstanceRepository;
-    
+
     @Test
     public void shouldGetReport() {
         // Given
-        final LocalDate start = LocalDate.now().withDayOfMonth(1);
+        final LocalDate start = LocalDate.of(2025, 1, 1);
         final LocalDate end = start.withDayOfMonth(start.getMonth().length(start.isLeapYear()));
         final ZoneId zoneId = ZoneId.systemDefault();
-        
+
         LocalDate from = start;
         int days = 0;
         // generate one month of service instance
-        
+
         while (from.toEpochDay() < end.toEpochDay()) {
             Instant createAt = from.atStartOfDay(zoneId).toInstant();
             Instant updatedAt = from.atStartOfDay(zoneId).plus(Duration.ofHours(10)).toInstant();
@@ -62,14 +62,14 @@ public abstract class AbstractServiceUsageReportTest {
             from = from.plusDays(1);
             days++;
         }
-        
-        
+
+
         // When
         Instant now = end.plusDays(1).atStartOfDay(zoneId).toInstant();
         ServiceUsageReport.ServiceUsageEvent event = serviceUsageReport.report(now,
             Reportable.TimeInterval.of(start.atStartOfDay(zoneId), end.plusDays(1).atStartOfDay(zoneId))
         );
-        
+
         // Then
         List<ServiceUsage.DailyServiceStatistics> statistics = event.services().dailyStatistics();
         Assertions.assertEquals(ServiceType.values().length - 1, statistics.size());

core/src/test/java/io/kestra/core/repositories/AbstractExecutionRepositoryTest.java

@@ -344,27 +344,6 @@ public abstract class AbstractExecutionRepositoryTest {
         assertThat(executions.getTotal()).isEqualTo(8L);
     }
 
-    @Test
-    protected void findTaskRun() {
-        var tenant = TestsUtils.randomTenant(this.getClass().getSimpleName());
-inject(tenant);
-
-        ArrayListTotal<TaskRun> taskRuns = executionRepository.findTaskRun(Pageable.from(1, 10), tenant, null);
-        assertThat(taskRuns.getTotal()).isEqualTo(74L);
-        assertThat(taskRuns.size()).isEqualTo(10);
-
-        var filters = List.of(QueryFilter.builder()
-            .field(QueryFilter.Field.LABELS)
-            .operation(QueryFilter.Op.EQUALS)
-            .value(Map.of("key", "value"))
-            .build());
-
-        taskRuns = executionRepository.findTaskRun(Pageable.from(1, 10), tenant, filters);
-        assertThat(taskRuns.getTotal()).isEqualTo(1L);
-        assertThat(taskRuns.size()).isEqualTo(1);
-    }
-
-
     @Test
     protected void findById() {
         var tenant = TestsUtils.randomTenant(this.getClass().getSimpleName());
@@ -408,6 +387,21 @@ inject(tenant);
         assertThat(full.isPresent()).isFalse();
     }
 
+    @Test
+    protected void purgeExecutions() {
+        var tenant = TestsUtils.randomTenant(this.getClass().getSimpleName());
+        var execution1 = ExecutionFixture.EXECUTION_1(tenant);
+        executionRepository.save(execution1);
+        var execution2 = ExecutionFixture.EXECUTION_2(tenant);
+        executionRepository.save(execution2);
+
+        var results = executionRepository.purge(List.of(execution1, execution2));
+        assertThat(results).isEqualTo(2);
+
+        assertThat(executionRepository.findById(tenant, execution1.getId())).isEmpty();
+        assertThat(executionRepository.findById(tenant, execution2.getId())).isEmpty();
+    }
+
     @Test
     protected void delete() {
         var tenant = TestsUtils.randomTenant(this.getClass().getSimpleName());
@@ -463,8 +457,7 @@ inject(tenant);
             ZonedDateTime.now().minusDays(10),
             ZonedDateTime.now(),
             null,
-            null,
-            false);
+            null);
 
         assertThat(result.size()).isEqualTo(11);
         assertThat(result.get(10).getExecutionCounts().size()).isEqualTo(11);
@@ -483,8 +476,7 @@ inject(tenant);
             ZonedDateTime.now().minusDays(10),
             ZonedDateTime.now(),
             null,
-            null,
-            false);
+            null);
 
         assertThat(result.size()).isEqualTo(11);
         assertThat(result.get(10).getExecutionCounts().get(State.Type.SUCCESS)).isEqualTo(21L);
@@ -498,8 +490,7 @@ inject(tenant);
             ZonedDateTime.now().minusDays(10),
             ZonedDateTime.now(),
             null,
-            null,
-            false);
+            null);
         assertThat(result.size()).isEqualTo(11);
         assertThat(result.get(10).getExecutionCounts().get(State.Type.SUCCESS)).isEqualTo(20L);
 
@@ -512,93 +503,11 @@ inject(tenant);
             ZonedDateTime.now().minusDays(10),
             ZonedDateTime.now(),
             null,
-            null,
-            false);
+            null);
         assertThat(result.size()).isEqualTo(11);
         assertThat(result.get(10).getExecutionCounts().get(State.Type.SUCCESS)).isEqualTo(1L);
     }
 
-    @Test
-    protected void taskRunsDailyStatistics() {
-        var tenant = TestsUtils.randomTenant(this.getClass().getSimpleName());
-        for (int i = 0; i < 28; i++) {
-            executionRepository.save(builder(
-                tenant,
-                i < 5 ? State.Type.RUNNING : (i < 8 ? State.Type.FAILED : State.Type.SUCCESS),
-                i < 15 ? null : "second"
-            ).build());
-        }
-
-        executionRepository.save(builder(
-            tenant,
-            State.Type.SUCCESS,
-            "second"
-        ).namespace(NamespaceUtils.SYSTEM_FLOWS_DEFAULT_NAMESPACE).build());
-
-        List<DailyExecutionStatistics> result = executionRepository.dailyStatistics(
-            null,
-            tenant,
-            null,
-            null,
-            null,
-            ZonedDateTime.now().minusDays(10),
-            ZonedDateTime.now(),
-            null,
-            null,
-            true);
-
-        assertThat(result.size()).isEqualTo(11);
-        assertThat(result.get(10).getExecutionCounts().size()).isEqualTo(11);
-        assertThat(result.get(10).getDuration().getAvg().toMillis()).isGreaterThan(0L);
-
-        assertThat(result.get(10).getExecutionCounts().get(State.Type.FAILED)).isEqualTo(3L * 2);
-        assertThat(result.get(10).getExecutionCounts().get(State.Type.RUNNING)).isEqualTo(5L * 2);
-        assertThat(result.get(10).getExecutionCounts().get(State.Type.SUCCESS)).isEqualTo(57L);
-
-        result = executionRepository.dailyStatistics(
-            null,
-            tenant,
-            List.of(FlowScope.USER, FlowScope.SYSTEM),
-            null,
-            null,
-            ZonedDateTime.now().minusDays(10),
-            ZonedDateTime.now(),
-            null,
-            null,
-            true);
-
-        assertThat(result.size()).isEqualTo(11);
-        assertThat(result.get(10).getExecutionCounts().get(State.Type.SUCCESS)).isEqualTo(57L);
-
-        result = executionRepository.dailyStatistics(
-            null,
-            tenant,
-            List.of(FlowScope.USER),
-            null,
-            null,
-            ZonedDateTime.now().minusDays(10),
-            ZonedDateTime.now(),
-            null,
-            null,
-            true);
-        assertThat(result.size()).isEqualTo(11);
-        assertThat(result.get(10).getExecutionCounts().get(State.Type.SUCCESS)).isEqualTo(55L);
-
-        result = executionRepository.dailyStatistics(
-            null,
-            tenant,
-            List.of(FlowScope.SYSTEM),
-            null,
-            null,
-            ZonedDateTime.now().minusDays(10),
-            ZonedDateTime.now(),
-            null,
-            null,
-            true);
-        assertThat(result.size()).isEqualTo(11);
-        assertThat(result.get(10).getExecutionCounts().get(State.Type.SUCCESS)).isEqualTo(2L);
-    }
-
     @SuppressWarnings("OptionalGetWithoutIsPresent")
     @Test
     protected void executionsCount() throws InterruptedException {
@@ -784,4 +693,91 @@ inject(tenant);
         assertThat(flowIds.size()).isEqualTo(lastExecutions.size());
     }
 
+    private static final Instant NOW = Instant.now();
+    private static final Instant INSTANT_ONE = NOW.minus(Duration.ofDays(1000));
+    private static final Instant INSTANT_TWO = INSTANT_ONE.plus(Duration.ofHours(11));
+    private static final Instant INSTANT_THREE = INSTANT_TWO.plus(Duration.ofHours(222));
+
+    @Test
+    protected void findShouldSortCorrectlyOnDurationAndDates() {
+        // given
+        var tenant = TestsUtils.randomTenant(this.getClass().getSimpleName());
+        var createdExecution = Execution.builder()
+            .id("createdExecution__"+FriendlyId.createFriendlyId())
+            .namespace(NAMESPACE)
+            .tenantId(tenant)
+            .flowId(FLOW)
+            .flowRevision(1)
+            .state(
+                State.of(
+                    State.Type.CREATED,
+                    List.of(
+                        new State.History(State.Type.CREATED, INSTANT_ONE)
+                    )
+                )
+            ).build();
+        assertThat(createdExecution.getState().getDuration()).isCloseTo(Duration.ofDays(1000), Duration.ofMinutes(10));
+        executionRepository.save(createdExecution);
+
+        var successExecution = Execution.builder()
+            .id("successExecution__"+FriendlyId.createFriendlyId())
+            .namespace(NAMESPACE)
+            .tenantId(tenant)
+            .flowId(FLOW)
+            .flowRevision(1)
+            .state(
+                State.of(
+                    State.Type.SUCCESS,
+                    List.of(
+                        new State.History(State.Type.CREATED, INSTANT_ONE),
+                        new State.History(State.Type.RUNNING, INSTANT_TWO),
+                        new State.History(State.Type.SUCCESS, INSTANT_THREE)
+                    )
+                )
+            ).build();
+        assertThat(successExecution.getState().getDuration()).isCloseTo(Duration.ofHours(233), Duration.ofMinutes(10));
+        executionRepository.save(successExecution);
+
+        var runningExecution = Execution.builder()
+            .id("runningExecution__"+FriendlyId.createFriendlyId())
+            .namespace(NAMESPACE)
+            .tenantId(tenant)
+            .flowId(FLOW)
+            .flowRevision(1)
+            .state(
+                State.of(
+                    State.Type.RUNNING,
+                    List.of(
+                        new State.History(State.Type.CREATED, INSTANT_TWO),
+                        new State.History(State.Type.RUNNING, INSTANT_THREE)
+                    )
+                )
+            ).build();
+        assertThat(runningExecution.getState().getDuration()).isCloseTo(Duration.ofDays(1000).minus(Duration.ofHours(11)), Duration.ofMinutes(10));
+        executionRepository.save(runningExecution);
+
+        // when
+        List<QueryFilter> emptyFilters = null;
+        var sortedByShortestDuration = executionRepository.find(Pageable.from(Sort.of(Sort.Order.asc("state_duration"))), tenant, emptyFilters);
+        // then
+        assertThat(sortedByShortestDuration.stream())
+            .as("assert order when finding by shortest duration")
+            .usingRecursiveFieldByFieldElementComparatorOnFields("id")
+            .containsExactly(
+                successExecution,
+                runningExecution,
+                createdExecution
+            );
+
+        // when
+        var findByMoreRecentStartDate = executionRepository.find(Pageable.from(1,1, Sort.of(Sort.Order.desc("start_date"))), tenant, emptyFilters);
+        // then
+        assertThat(findByMoreRecentStartDate.stream())
+            .as("assert order when finding by last start date")
+            .usingRecursiveFieldByFieldElementComparatorOnFields("id")
+            .containsExactly(
+                runningExecution
+            );
+    }
+
 }

core/src/test/java/io/kestra/core/repositories/AbstractExecutionServiceTest.java

@@ -114,7 +114,8 @@ public abstract class AbstractExecutionServiceTest {
             flow.getId(),
             null,
             ZonedDateTime.now(),
-            null
+            null,
+            100
         );
 
         assertThat(purge.getExecutionsCount()).isEqualTo(1);
@@ -132,7 +133,8 @@ public abstract class AbstractExecutionServiceTest {
             flow.getId(),
             null,
             ZonedDateTime.now(),
-            null
+            null,
+            100
         );
 
         assertThat(purge.getExecutionsCount()).isZero();

core/src/test/java/io/kestra/core/repositories/AbstractFlowRepositoryTest.java

@@ -32,7 +32,6 @@ import jakarta.validation.ConstraintViolationException;
 import java.util.concurrent.CopyOnWriteArrayList;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
-import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -48,7 +47,9 @@ import java.util.stream.Stream;
 import static io.kestra.core.models.flows.FlowScope.SYSTEM;
 import static io.kestra.core.utils.NamespaceUtils.SYSTEM_FLOWS_DEFAULT_NAMESPACE;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 @KestraTest
 public abstract class AbstractFlowRepositoryTest {
@@ -644,7 +645,7 @@ public abstract class AbstractFlowRepositoryTest {
             int count = flowRepository.count(tenant);
 
             // Then
-            Assertions.assertTrue(count > 0);
+            assertTrue(count > 0);
         } finally {
             Optional.ofNullable(toDelete).ifPresent(flow -> {
                 flowRepository.delete(flow);
@@ -652,6 +653,36 @@ public abstract class AbstractFlowRepositoryTest {
         }
     }
 
+    @Test
+    void should_exist_for_tenant(){
+        String tenantFlowExist = TestsUtils.randomTenant(this.getClass().getSimpleName());
+        FlowWithSource flowExist = FlowWithSource.builder()
+            .id("flowExist")
+            .namespace(SYSTEM_FLOWS_DEFAULT_NAMESPACE)
+            .tenantId(tenantFlowExist)
+            .deleted(false)
+            .build();
+        flowExist = flowRepository.create(GenericFlow.of(flowExist));
+
+        String tenantFlowDeleted = TestsUtils.randomTenant(this.getClass().getSimpleName());
+        FlowWithSource flowDeleted = FlowWithSource.builder()
+            .id("flowDeleted")
+            .namespace(SYSTEM_FLOWS_DEFAULT_NAMESPACE)
+            .tenantId(tenantFlowDeleted)
+            .deleted(true)
+            .build();
+        flowDeleted = flowRepository.create(GenericFlow.of(flowDeleted));
+
+        try {
+            assertTrue(flowRepository.existAnyNoAcl(tenantFlowExist));
+            assertFalse(flowRepository.existAnyNoAcl("not_found"));
+            assertFalse(flowRepository.existAnyNoAcl(tenantFlowDeleted));
+        } finally {
+            deleteFlow(flowExist);
+            deleteFlow(flowDeleted);
+        }
+    }
+
     private static Flow createTestFlowForNamespace(String tenantId, String namespace) {
         return Flow.builder()
             .id(IdUtils.create())

core/src/test/java/io/kestra/core/repositories/AbstractLoadedFlowRepositoryTest.java

@@ -1,281 +0,0 @@
-package io.kestra.core.repositories;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-import io.kestra.core.Helpers;
-import io.kestra.core.junit.annotations.KestraTest;
-import io.kestra.core.models.QueryFilter;
-import io.kestra.core.models.SearchResult;
-import io.kestra.core.models.flows.Flow;
-import io.kestra.core.models.flows.FlowWithSource;
-import io.kestra.core.utils.TestsUtils;
-import io.micronaut.data.model.Pageable;
-import io.micronaut.data.model.Sort;
-import jakarta.inject.Inject;
-import java.io.IOException;
-import java.net.URISyntaxException;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicBoolean;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-@KestraTest
-public abstract class AbstractLoadedFlowRepositoryTest {
-
-    @Inject
-    protected FlowRepositoryInterface flowRepository;
-
-    @Inject
-    protected ExecutionRepositoryInterface executionRepository;
-
-    @Inject
-    private LocalFlowRepositoryLoader repositoryLoader;
-
-    protected static final String TENANT = TestsUtils.randomTenant(AbstractLoadedFlowRepositoryTest.class.getSimpleName());
-    private static final AtomicBoolean IS_INIT = new AtomicBoolean();
-
-    @BeforeEach
-    protected synchronized void init() throws IOException, URISyntaxException {
-        initFlows(repositoryLoader);
-    }
-
-    protected static synchronized void initFlows(LocalFlowRepositoryLoader repo) throws IOException, URISyntaxException {
-        if (!IS_INIT.get()){
-            TestsUtils.loads(TENANT, repo);
-            IS_INIT.set(true);
-        }
-    }
-
-    @Test
-    void findAll() {
-        List<Flow> save = flowRepository.findAll(TENANT);
-
-        assertThat((long) save.size()).isEqualTo(Helpers.FLOWS_COUNT);
-    }
-
-    @Test
-    void findAllWithSource() {
-        List<FlowWithSource> save = flowRepository.findAllWithSource(TENANT);
-
-        assertThat((long) save.size()).isEqualTo(Helpers.FLOWS_COUNT);
-    }
-
-    @Test
-    void findAllForAllTenants() {
-        List<Flow> save = flowRepository.findAllForAllTenants();
-
-        assertThat((long) save.size()).isEqualTo(Helpers.FLOWS_COUNT);
-    }
-
-    @Test
-    void findAllWithSourceForAllTenants() {
-        List<FlowWithSource> save = flowRepository.findAllWithSourceForAllTenants();
-
-        assertThat((long) save.size()).isEqualTo(Helpers.FLOWS_COUNT);
-    }
-
-    @Test
-    void findByNamespace() {
-        List<Flow> save = flowRepository.findByNamespace(TENANT, "io.kestra.tests");
-        assertThat((long) save.size()).isEqualTo(Helpers.FLOWS_COUNT - 24);
-
-        save = flowRepository.findByNamespace(TENANT, "io.kestra.tests2");
-        assertThat((long) save.size()).isEqualTo(1L);
-
-        save = flowRepository.findByNamespace(TENANT, "io.kestra.tests.minimal.bis");
-        assertThat((long) save.size()).isEqualTo(1L);
-    }
-
-    @Test
-    void findByNamespacePrefix() {
-        List<Flow> save = flowRepository.findByNamespacePrefix(TENANT, "io.kestra.tests");
-        assertThat((long) save.size()).isEqualTo(Helpers.FLOWS_COUNT - 1);
-
-        save = flowRepository.findByNamespace(TENANT, "io.kestra.tests2");
-        assertThat((long) save.size()).isEqualTo(1L);
-
-        save = flowRepository.findByNamespace(TENANT, "io.kestra.tests.minimal.bis");
-        assertThat((long) save.size()).isEqualTo(1L);
-    }
-
-    @Test
-    void findByNamespacePrefixWithSource() {
-        List<FlowWithSource> save = flowRepository.findByNamespacePrefixWithSource(TENANT, "io.kestra.tests");
-        assertThat((long) save.size()).isEqualTo(Helpers.FLOWS_COUNT - 1);
-    }
-
-    @Test
-    void find_paginationPartial() {
-        assertThat(flowRepository.find(Pageable.from(1, (int) Helpers.FLOWS_COUNT - 1, Sort.UNSORTED), TENANT, null)
-            .size())
-            .describedAs("When paginating at MAX-1, it should return MAX-1")
-            .isEqualTo(Helpers.FLOWS_COUNT - 1);
-
-        assertThat(flowRepository.findWithSource(Pageable.from(1, (int) Helpers.FLOWS_COUNT - 1, Sort.UNSORTED), TENANT, null)
-            .size())
-            .describedAs("When paginating at MAX-1, it should return MAX-1")
-            .isEqualTo(Helpers.FLOWS_COUNT - 1);
-    }
-
-    @Test
-    void find_paginationGreaterThanExisting() {
-        assertThat(flowRepository.find(Pageable.from(1, (int) Helpers.FLOWS_COUNT + 1, Sort.UNSORTED), TENANT, null)
-            .size())
-            .describedAs("When paginating requesting a larger amount than existing, it should return existing MAX")
-            .isEqualTo(Helpers.FLOWS_COUNT);
-        assertThat(flowRepository.findWithSource(Pageable.from(1, (int) Helpers.FLOWS_COUNT + 1, Sort.UNSORTED), TENANT, null)
-            .size())
-            .describedAs("When paginating requesting a larger amount than existing, it should return existing MAX")
-            .isEqualTo(Helpers.FLOWS_COUNT);
-    }
-
-    @Test
-    void find_prefixMatchingAllNamespaces() {
-        assertThat(flowRepository.find(
-            Pageable.UNPAGED,
-            TENANT,
-            List.of(
-                QueryFilter.builder().field(QueryFilter.Field.NAMESPACE).operation(QueryFilter.Op.STARTS_WITH).value("io.kestra.tests").build()
-            )
-        ).size())
-            .describedAs("When filtering on NAMESPACE START_WITH a pattern that match all, it should return all")
-            .isEqualTo(Helpers.FLOWS_COUNT);
-
-        assertThat(flowRepository.findWithSource(
-            Pageable.UNPAGED,
-            TENANT,
-            List.of(
-                QueryFilter.builder().field(QueryFilter.Field.NAMESPACE).operation(QueryFilter.Op.STARTS_WITH).value("io.kestra.tests").build()
-            )
-        ).size())
-            .describedAs("When filtering on NAMESPACE START_WITH a pattern that match all, it should return all")
-            .isEqualTo(Helpers.FLOWS_COUNT);
-    }
-
-    @Test
-    void find_aSpecifiedNamespace() {
-        assertThat(flowRepository.find(
-            Pageable.UNPAGED,
-            TENANT,
-            List.of(
-                QueryFilter.builder().field(QueryFilter.Field.NAMESPACE).operation(QueryFilter.Op.EQUALS).value("io.kestra.tests2").build()
-            )
-        ).size()).isEqualTo(1L);
-
-        assertThat(flowRepository.findWithSource(
-            Pageable.UNPAGED,
-            TENANT,
-            List.of(
-                QueryFilter.builder().field(QueryFilter.Field.NAMESPACE).operation(QueryFilter.Op.EQUALS).value("io.kestra.tests2").build()
-            )
-        ).size()).isEqualTo(1L);
-    }
-
-    @Test
-    void find_aSpecificSubNamespace() {
-        assertThat(flowRepository.find(
-            Pageable.UNPAGED,
-            TENANT,
-            List.of(
-                QueryFilter.builder().field(QueryFilter.Field.NAMESPACE).operation(QueryFilter.Op.EQUALS).value("io.kestra.tests.minimal.bis").build()
-            )
-        ).size())
-            .isEqualTo(1L);
-
-        assertThat(flowRepository.findWithSource(
-            Pageable.UNPAGED,
-            TENANT,
-            List.of(
-                QueryFilter.builder().field(QueryFilter.Field.NAMESPACE).operation(QueryFilter.Op.EQUALS).value("io.kestra.tests.minimal.bis").build()
-            )
-        ).size())
-            .isEqualTo(1L);
-    }
-
-    @Test
-    void find_aSpecificLabel() {
-        assertThat(
-            flowRepository.find(Pageable.UNPAGED, TENANT,
-                List.of(
-                    QueryFilter.builder().field(QueryFilter.Field.LABELS).operation(QueryFilter.Op.EQUALS).value(
-                        Map.of("country", "FR")).build()
-                )
-            ).size())
-            .isEqualTo(1);
-
-        assertThat(
-            flowRepository.findWithSource(Pageable.UNPAGED, TENANT,
-                List.of(
-                    QueryFilter.builder().field(QueryFilter.Field.LABELS).operation(QueryFilter.Op.EQUALS).value(Map.of("country", "FR")).build()
-                )
-            ).size())
-            .isEqualTo(1);
-    }
-
-    @Test
-    void find_aSpecificFlowByNamespaceAndLabel() {
-        assertThat(
-            flowRepository.find(Pageable.UNPAGED, TENANT,
-                List.of(
-                    QueryFilter.builder().field(QueryFilter.Field.NAMESPACE).operation(QueryFilter.Op.EQUALS).value("io.kestra.tests").build(),
-                    QueryFilter.builder().field(QueryFilter.Field.LABELS).operation(QueryFilter.Op.EQUALS).value(Map.of("key2", "value2")).build()
-                )
-            ).size())
-            .isEqualTo(1);
-
-        assertThat(
-            flowRepository.findWithSource(Pageable.UNPAGED, TENANT,
-                List.of(
-                    QueryFilter.builder().field(QueryFilter.Field.NAMESPACE).operation(QueryFilter.Op.EQUALS).value("io.kestra.tests").build(),
-                    QueryFilter.builder().field(QueryFilter.Field.LABELS).operation(QueryFilter.Op.EQUALS).value(Map.of("key2", "value2")).build()
-                )
-            ).size())
-            .isEqualTo(1);
-    }
-
-    @Test
-    void find_noResult_forAnUnknownNamespace() {
-        assertThat(
-            flowRepository.find(Pageable.UNPAGED, TENANT,
-                List.of(
-                    QueryFilter.builder().field(QueryFilter.Field.NAMESPACE).operation(QueryFilter.Op.EQUALS).value("io.kestra.tests").build(),
-                    QueryFilter.builder().field(QueryFilter.Field.LABELS).operation(QueryFilter.Op.EQUALS).value(Map.of("key1", "value2")).build()
-                )
-            ).size())
-            .isEqualTo(0);
-
-        assertThat(
-            flowRepository.findWithSource(Pageable.UNPAGED, TENANT,
-                List.of(
-                    QueryFilter.builder().field(QueryFilter.Field.NAMESPACE).operation(QueryFilter.Op.EQUALS).value("io.kestra.tests").build(),
-                    QueryFilter.builder().field(QueryFilter.Field.LABELS).operation(QueryFilter.Op.EQUALS).value(Map.of("key1", "value2")).build()
-                )
-            ).size())
-            .isEqualTo(0);
-    }
-
-    @Test
-    protected void findSpecialChars() {
-        ArrayListTotal<SearchResult<Flow>> save = flowRepository.findSourceCode(Pageable.unpaged(), "https://api.chucknorris.io", TENANT, null);
-        assertThat((long) save.size()).isEqualTo(2L);
-    }
-
-
-    @Test
-    void findDistinctNamespace() {
-        List<String> distinctNamespace = flowRepository.findDistinctNamespace(TENANT);
-        assertThat((long) distinctNamespace.size()).isEqualTo(9L);
-    }
-
-    @Test
-    void shouldReturnForGivenQueryWildCardFilters() {
-        List<QueryFilter> filters = List.of(
-            QueryFilter.builder().field(QueryFilter.Field.QUERY).operation(QueryFilter.Op.EQUALS).value("*").build()
-        );
-        ArrayListTotal<Flow> flows = flowRepository.find(Pageable.from(1, 10), TENANT, filters);
-        assertThat(flows.size()).isEqualTo(10);
-        assertThat(flows.getTotal()).isEqualTo(Helpers.FLOWS_COUNT);
-    }
-
-}

core/src/test/java/io/kestra/core/repositories/AbstractLogRepositoryTest.java

@@ -14,6 +14,7 @@ import io.kestra.core.models.flows.State;
 import io.kestra.core.repositories.ExecutionRepositoryInterface.ChildFilter;
 import io.kestra.core.utils.IdUtils;
 import io.kestra.core.utils.TestsUtils;
+import io.kestra.plugin.core.dashboard.data.Executions;
 import io.kestra.plugin.core.dashboard.data.Logs;
 import io.micronaut.data.model.Pageable;
 import jakarta.inject.Inject;
@@ -359,4 +360,16 @@ public abstract class AbstractLogRepositoryTest {
 
         assertThat(results).hasSize(1);
     }
+
+    @Test
+    void purge() {
+        String tenant = TestsUtils.randomTenant(this.getClass().getSimpleName());
+        logRepository.save(logEntry(tenant, Level.INFO, "execution1").build());
+        logRepository.save(logEntry(tenant, Level.INFO, "execution1").build());
+        logRepository.save(logEntry(tenant, Level.INFO, "execution2").build());
+        logRepository.save(logEntry(tenant, Level.INFO, "execution2").build());
+
+        var result = logRepository.purge(List.of(Execution.builder().id("execution1").build(), Execution.builder().id("execution2").build()));
+        assertThat(result).isEqualTo(4);
+    }
 }

core/src/test/java/io/kestra/core/repositories/AbstractMetricRepositoryTest.java

@@ -1,6 +1,7 @@
 package io.kestra.core.repositories;
 
 import com.devskiller.friendly_id.FriendlyId;
+import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.ExecutionKind;
 import io.kestra.core.models.executions.MetricEntry;
 import io.kestra.core.models.executions.TaskRun;
@@ -12,6 +13,7 @@ import io.micronaut.data.model.Pageable;
 import io.kestra.core.junit.annotations.KestraTest;
 import jakarta.inject.Inject;
 import org.junit.jupiter.api.Test;
+import org.slf4j.event.Level;
 
 import java.time.Duration;
 import java.time.ZonedDateTime;
@@ -119,6 +121,18 @@ public abstract class AbstractMetricRepositoryTest {
         assertThat(results).hasSize(3);
     }
 
+    @Test
+    void purge() {
+        String tenant = TestsUtils.randomTenant(this.getClass().getSimpleName());
+        metricRepository.save(MetricEntry.of(taskRun(tenant, "execution1", "task"), counter("counter1"), null));
+        metricRepository.save(MetricEntry.of(taskRun(tenant, "execution1", "task"), counter("counter2"), null));
+        metricRepository.save(MetricEntry.of(taskRun(tenant, "execution2", "task"), counter("counter1"), null));
+        metricRepository.save(MetricEntry.of(taskRun(tenant, "execution2", "task"), counter("counter2"), null));
+
+        var result = metricRepository.purge(List.of(Execution.builder().id("execution1").build(), Execution.builder().id("execution2").build()));
+        assertThat(result).isEqualTo(4);
+    }
+
     private Counter counter(String metricName) {
         return Counter.of(metricName, 1);
     }

core/src/test/java/io/kestra/core/runners/AbstractRunnerTest.java

@@ -1,7 +1,6 @@
 package io.kestra.core.runners;
 
 import io.kestra.core.junit.annotations.ExecuteFlow;
-import io.kestra.core.junit.annotations.FlakyTest;
 import io.kestra.core.junit.annotations.KestraTest;
 import io.kestra.core.junit.annotations.LoadFlows;
 import io.kestra.core.models.executions.Execution;
@@ -29,15 +28,17 @@ import static org.assertj.core.api.Assertions.assertThat;
 // must be per-class to allow calling once init() which took a lot of time
 public abstract class AbstractRunnerTest {
 
+    public static final String TENANT_1 = "tenant1";
+    public static final String TENANT_2 = "tenant2";
     @Inject
-    protected RunnerUtils runnerUtils;
+    protected TestRunnerUtils runnerUtils;
 
     @Inject
     @Named(QueueFactoryInterface.WORKERTASKLOG_NAMED)
     protected QueueInterface<LogEntry> logsQueue;
 
     @Inject
-    private RestartCaseTest restartCaseTest;
+    protected RestartCaseTest restartCaseTest;
 
     @Inject
     protected FlowTriggerCaseTest flowTriggerCaseTest;
@@ -49,13 +50,13 @@ public abstract class AbstractRunnerTest {
     private PluginDefaultsCaseTest pluginDefaultsCaseTest;
 
     @Inject
-    private FlowCaseTest flowCaseTest;
+    protected FlowCaseTest flowCaseTest;
 
     @Inject
     private WorkingDirectoryTest.Suite workingDirectoryTest;
 
     @Inject
-    private PauseTest.Suite pauseTest;
+    protected PauseTest.Suite pauseTest;
 
     @Inject
     private SkipExecutionCaseTest skipExecutionCaseTest;
@@ -67,10 +68,10 @@ public abstract class AbstractRunnerTest {
     protected LoopUntilCaseTest loopUntilTestCaseTest;
 
     @Inject
-    private FlowConcurrencyCaseTest flowConcurrencyCaseTest;
+    protected FlowConcurrencyCaseTest flowConcurrencyCaseTest;
 
     @Inject
-    private ScheduleDateCaseTest scheduleDateCaseTest;
+    protected ScheduleDateCaseTest scheduleDateCaseTest;
 
     @Inject
     protected FlowInputOutput flowIO;
@@ -79,7 +80,7 @@ public abstract class AbstractRunnerTest {
     private SLATestCase slaTestCase;
 
     @Inject
-    private ChangeStateTestCase changeStateTestCase;
+    protected ChangeStateTestCase changeStateTestCase;
 
     @Inject
     private AfterExecutionTestCase afterExecutionTestCase;
@@ -172,7 +173,7 @@ public abstract class AbstractRunnerTest {
 
     @Test
     @LoadFlows({"flows/valids/restart_local_errors.yaml"})
-    void restartFailedThenFailureWithLocalErrors() throws Exception {
+    protected void restartFailedThenFailureWithLocalErrors() throws Exception {
         restartCaseTest.restartFailedThenFailureWithLocalErrors();
     }
 
@@ -195,12 +196,12 @@ public abstract class AbstractRunnerTest {
     }
 
     @Test
-    @LoadFlows({"flows/valids/trigger-flow-listener-no-inputs.yaml",
+    @LoadFlows(value = {"flows/valids/trigger-flow-listener-no-inputs.yaml",
         "flows/valids/trigger-flow-listener.yaml",
         "flows/valids/trigger-flow-listener-namespace-condition.yaml",
-        "flows/valids/trigger-flow.yaml"})
+        "flows/valids/trigger-flow.yaml"}, tenantId = "listener-tenant")
     void flowTrigger() throws Exception {
-        flowTriggerCaseTest.trigger();
+        flowTriggerCaseTest.trigger("listener-tenant");
     }
 
     @Test // flaky on CI but never fail locally
@@ -210,13 +211,11 @@ public abstract class AbstractRunnerTest {
         flowTriggerCaseTest.triggerWithPause();
     }
 
-    @FlakyTest
-    @Disabled
     @Test
-    @LoadFlows({"flows/valids/trigger-flow-listener-with-concurrency-limit.yaml",
-        "flows/valids/trigger-flow-with-concurrency-limit.yaml"})
-    void flowTriggerWithConcurrencyLimit() throws Exception {
-        flowTriggerCaseTest.triggerWithConcurrencyLimit();
+    @LoadFlows(value = {"flows/valids/trigger-flow-listener-with-concurrency-limit.yaml",
+        "flows/valids/trigger-flow-with-concurrency-limit.yaml"}, tenantId = "trigger-tenant")
+    protected void flowTriggerWithConcurrencyLimit() throws Exception {
+        flowTriggerCaseTest.triggerWithConcurrencyLimit("trigger-tenant");
     }
 
     @Test
@@ -228,11 +227,11 @@ public abstract class AbstractRunnerTest {
     }
 
     @Test // Flaky on CI but never locally even with 100 repetitions
-    @LoadFlows({"flows/valids/trigger-flow-listener-namespace-condition.yaml",
+    @LoadFlows(value = {"flows/valids/trigger-flow-listener-namespace-condition.yaml",
         "flows/valids/trigger-multiplecondition-flow-c.yaml",
-        "flows/valids/trigger-multiplecondition-flow-d.yaml"})
+        "flows/valids/trigger-multiplecondition-flow-d.yaml"}, tenantId = "condition-tenant")
     void multipleConditionTriggerFailed() throws Exception {
-        multipleConditionTriggerCaseTest.failed();
+        multipleConditionTriggerCaseTest.failed("condition-tenant");
     }
 
     @Test
@@ -245,11 +244,11 @@ public abstract class AbstractRunnerTest {
 
     @Disabled
     @Test
-    @LoadFlows({"flows/valids/flow-trigger-preconditions-flow-listen.yaml",
+    @LoadFlows(value = {"flows/valids/flow-trigger-preconditions-flow-listen.yaml",
         "flows/valids/flow-trigger-preconditions-flow-a.yaml",
-        "flows/valids/flow-trigger-preconditions-flow-b.yaml"})
+        "flows/valids/flow-trigger-preconditions-flow-b.yaml"}, tenantId = TENANT_1)
     void flowTriggerPreconditionsMergeOutputs() throws Exception {
-        multipleConditionTriggerCaseTest.flowTriggerPreconditionsMergeOutputs();
+        multipleConditionTriggerCaseTest.flowTriggerPreconditionsMergeOutputs(TENANT_1);
     }
 
     @Test
@@ -258,6 +257,12 @@ public abstract class AbstractRunnerTest {
         multipleConditionTriggerCaseTest.flowTriggerOnPaused();
     }
 
+    @Test
+    @LoadFlows({"flows/valids/flow-trigger-for-each-item-parent.yaml", "flows/valids/flow-trigger-for-each-item-child.yaml", "flows/valids/flow-trigger-for-each-item-grandchild.yaml"})
+    void forEachItemWithFlowTrigger() throws Exception {
+        multipleConditionTriggerCaseTest.forEachItemWithFlowTrigger();
+    }
+
     @Test
     @LoadFlows({"flows/valids/each-null.yaml"})
     void eachWithNull() throws Exception {
@@ -274,24 +279,24 @@ public abstract class AbstractRunnerTest {
     @LoadFlows({"flows/valids/switch.yaml",
         "flows/valids/task-flow.yaml",
         "flows/valids/task-flow-inherited-labels.yaml"})
-    void flowWaitSuccess() throws Exception {
+    protected void flowWaitSuccess() throws Exception {
         flowCaseTest.waitSuccess();
     }
 
     @Test
-    @LoadFlows({"flows/valids/switch.yaml",
+    @LoadFlows(value = {"flows/valids/switch.yaml",
         "flows/valids/task-flow.yaml",
-        "flows/valids/task-flow-inherited-labels.yaml"})
+        "flows/valids/task-flow-inherited-labels.yaml"}, tenantId = TENANT_1)
     void flowWaitFailed() throws Exception {
-        flowCaseTest.waitFailed();
+        flowCaseTest.waitFailed(TENANT_1);
     }
 
     @Test
-    @LoadFlows({"flows/valids/switch.yaml",
+    @LoadFlows(value = {"flows/valids/switch.yaml",
         "flows/valids/task-flow.yaml",
-        "flows/valids/task-flow-inherited-labels.yaml"})
+        "flows/valids/task-flow-inherited-labels.yaml"}, tenantId = TENANT_2)
     public void invalidOutputs() throws Exception {
-        flowCaseTest.invalidOutputs();
+        flowCaseTest.invalidOutputs(TENANT_2);
     }
 
     @Test
@@ -301,9 +306,9 @@ public abstract class AbstractRunnerTest {
     }
 
     @Test
-    @LoadFlows(value = {"flows/valids/working-directory.yaml"}, tenantId = "tenant1")
+    @LoadFlows(value = {"flows/valids/working-directory.yaml"}, tenantId = TENANT_1)
     public void workerFailed() throws Exception {
-        workingDirectoryTest.failed("tenant1", runnerUtils);
+        workingDirectoryTest.failed(TENANT_1, runnerUtils);
     }
 
     @Test
@@ -354,7 +359,6 @@ public abstract class AbstractRunnerTest {
         skipExecutionCaseTest.skipExecution();
     }
 
-    @Disabled
     @Test
     @LoadFlows({"flows/valids/for-each-item-subflow.yaml",
         "flows/valids/for-each-item.yaml"})
@@ -363,12 +367,11 @@ public abstract class AbstractRunnerTest {
     }
 
     @Test
-    @LoadFlows({"flows/valids/for-each-item.yaml"})
+    @LoadFlows(value = {"flows/valids/for-each-item.yaml"}, tenantId = TENANT_1)
     protected void forEachItemEmptyItems() throws Exception {
-        forEachItemCaseTest.forEachItemEmptyItems();
+        forEachItemCaseTest.forEachItemEmptyItems(TENANT_1);
     }
 
-    @Disabled
     @Test
     @LoadFlows({"flows/valids/for-each-item-subflow-failed.yaml",
         "flows/valids/for-each-item-failed.yaml"})
@@ -384,16 +387,16 @@ public abstract class AbstractRunnerTest {
     }
 
     @Test // flaky on CI but always pass locally even with 100 iterations
-    @LoadFlows({"flows/valids/restart-for-each-item.yaml", "flows/valids/restart-child.yaml"})
+    @LoadFlows(value = {"flows/valids/restart-for-each-item.yaml", "flows/valids/restart-child.yaml"}, tenantId = TENANT_1)
     void restartForEachItem() throws Exception {
-        forEachItemCaseTest.restartForEachItem();
+        forEachItemCaseTest.restartForEachItem(TENANT_1);
     }
 
     @Test
-    @LoadFlows({"flows/valids/for-each-item-subflow.yaml",
-        "flows/valids/for-each-item-in-if.yaml"})
+    @LoadFlows(value = {"flows/valids/for-each-item-subflow.yaml",
+        "flows/valids/for-each-item-in-if.yaml"}, tenantId = TENANT_1)
     protected void forEachItemInIf() throws Exception {
-        forEachItemCaseTest.forEachItemInIf();
+        forEachItemCaseTest.forEachItemInIf(TENANT_1);
     }
 
     @Test
@@ -434,12 +437,11 @@ public abstract class AbstractRunnerTest {
     }
 
     @Test
-    @LoadFlows({"flows/valids/flow-concurrency-for-each-item.yaml", "flows/valids/flow-concurrency-queue.yml"})
+    @LoadFlows(value = {"flows/valids/flow-concurrency-for-each-item.yaml", "flows/valids/flow-concurrency-queue.yml"}, tenantId = TENANT_1)
     protected void flowConcurrencyWithForEachItem() throws Exception {
-        flowConcurrencyCaseTest.flowConcurrencyWithForEachItem();
+        flowConcurrencyCaseTest.flowConcurrencyWithForEachItem(TENANT_1);
     }
 
-    @Disabled
     @Test
     @LoadFlows({"flows/valids/flow-concurrency-queue-fail.yml"})
     protected void concurrencyQueueRestarted() throws Exception {
@@ -453,9 +455,15 @@ public abstract class AbstractRunnerTest {
     }
 
     @Test
-    @LoadFlows({"flows/valids/flow-concurrency-subflow.yml", "flows/valids/flow-concurrency-cancel.yml"})
+    @LoadFlows(value = {"flows/valids/flow-concurrency-subflow.yml", "flows/valids/flow-concurrency-cancel.yml"}, tenantId = TENANT_1)
     void flowConcurrencySubflow() throws Exception {
-        flowConcurrencyCaseTest.flowConcurrencySubflow();
+        flowConcurrencyCaseTest.flowConcurrencySubflow(TENANT_1);
+    }
+
+    @Test
+    @LoadFlows({"flows/valids/flow-concurrency-parallel-subflow-kill.yaml", "flows/valids/flow-concurrency-parallel-subflow-kill-child.yaml", "flows/valids/flow-concurrency-parallel-subflow-kill-grandchild.yaml"})
+    void flowConcurrencyParallelSubflowKill() throws Exception {
+        flowConcurrencyCaseTest.flowConcurrencyParallelSubflowKill();
     }
 
     @Test
@@ -510,9 +518,9 @@ public abstract class AbstractRunnerTest {
     }
 
     @Test
-    @LoadFlows({"flows/valids/minimal.yaml"})
+    @LoadFlows(value = {"flows/valids/minimal.yaml"}, tenantId = TENANT_1)
     void shouldScheduleOnDate() throws Exception {
-        scheduleDateCaseTest.shouldScheduleOnDate();
+        scheduleDateCaseTest.shouldScheduleOnDate(TENANT_1);
     }
 
     @Test
@@ -534,15 +542,15 @@ public abstract class AbstractRunnerTest {
     }
 
     @Test
-    @LoadFlows({"flows/valids/sla-execution-condition.yaml"})
+    @LoadFlows(value = {"flows/valids/sla-execution-condition.yaml"}, tenantId = TENANT_1)
     void executionConditionSLAShouldCancel() throws Exception {
-        slaTestCase.executionConditionSLAShouldCancel();
+        slaTestCase.executionConditionSLAShouldCancel(TENANT_1);
     }
 
     @Test
-    @LoadFlows({"flows/valids/sla-execution-condition.yaml"})
+    @LoadFlows(value = {"flows/valids/sla-execution-condition.yaml"}, tenantId = TENANT_2)
     void executionConditionSLAShouldLabel() throws Exception {
-        slaTestCase.executionConditionSLAShouldLabel();
+        slaTestCase.executionConditionSLAShouldLabel(TENANT_2);
     }
 
     @Test
@@ -562,15 +570,15 @@ public abstract class AbstractRunnerTest {
     }
 
     @Test
-    @ExecuteFlow("flows/valids/failed-first.yaml")
+    @ExecuteFlow(value = "flows/valids/failed-first.yaml", tenantId = TENANT_1)
     public void changeStateShouldEndsInSuccess(Execution execution) throws Exception {
         changeStateTestCase.changeStateShouldEndsInSuccess(execution);
     }
 
     @Test
-    @LoadFlows({"flows/valids/failed-first.yaml", "flows/valids/subflow-parent-of-failed.yaml"})
+    @LoadFlows(value = {"flows/valids/failed-first.yaml", "flows/valids/subflow-parent-of-failed.yaml"}, tenantId = TENANT_2)
     public void changeStateInSubflowShouldEndsParentFlowInSuccess() throws Exception {
-        changeStateTestCase.changeStateInSubflowShouldEndsParentFlowInSuccess();
+        changeStateTestCase.changeStateInSubflowShouldEndsParentFlowInSuccess(TENANT_2);
     }
 
     @Test

core/src/test/java/io/kestra/core/runners/ChangeStateTestCase.java

@@ -3,25 +3,18 @@ package io.kestra.core.runners;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.State;
-import io.kestra.core.queues.QueueFactoryInterface;
-import io.kestra.core.queues.QueueInterface;
+import io.kestra.core.models.flows.State.Type;
 import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.services.ExecutionService;
-import io.kestra.core.utils.TestsUtils;
 import jakarta.inject.Inject;
-import jakarta.inject.Named;
 import jakarta.inject.Singleton;
-import reactor.core.publisher.Flux;
 
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
-
-import static io.kestra.core.tenant.TenantService.MAIN_TENANT;
 import static org.assertj.core.api.Assertions.assertThat;
 
 @Singleton
 public class ChangeStateTestCase {
+
+    public static final String NAMESPACE = "io.kestra.tests";
     @Inject
     private FlowRepositoryInterface flowRepository;
 
@@ -29,11 +22,7 @@ public class ChangeStateTestCase {
     private ExecutionService executionService;
 
     @Inject
-    @Named(QueueFactoryInterface.EXECUTION_NAMED)
-    private QueueInterface<Execution> executionQueue;
-
-    @Inject
-    private RunnerUtils runnerUtils;
+    private TestRunnerUtils runnerUtils;
 
     public void changeStateShouldEndsInSuccess(Execution execution) throws Exception {
         assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.FAILED);
@@ -41,73 +30,40 @@ public class ChangeStateTestCase {
         assertThat(execution.getTaskRunList().getFirst().getState().getCurrent()).isEqualTo(State.Type.FAILED);
 
         // await for the last execution
-        CountDownLatch latch = new CountDownLatch(1);
-        AtomicReference<Execution> lastExecution = new AtomicReference<>();
-        Flux<Execution> receivedExecutions = TestsUtils.receive(executionQueue, either -> {
-            Execution exec = either.getLeft();
-            if (execution.getId().equals(exec.getId()) && exec.getState().getCurrent() == State.Type.SUCCESS) {
-                lastExecution.set(exec);
-                latch.countDown();
-            }
-        });
-
         Flow flow = flowRepository.findByExecution(execution);
         Execution markedAs = executionService.markAs(execution, flow, execution.getTaskRunList().getFirst().getId(), State.Type.SUCCESS);
-        executionQueue.emit(markedAs);
+        Execution lastExecution = runnerUtils.emitAndAwaitExecution(e -> e.getState().getCurrent().equals(Type.SUCCESS), markedAs);
 
-        assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
-        receivedExecutions.blockLast();
-        assertThat(lastExecution.get().getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
-        assertThat(lastExecution.get().getTaskRunList()).hasSize(2);
-        assertThat(lastExecution.get().getTaskRunList().getFirst().getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+        assertThat(lastExecution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+        assertThat(lastExecution.getTaskRunList()).hasSize(2);
+        assertThat(lastExecution.getTaskRunList().getFirst().getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
     }
 
-    public void changeStateInSubflowShouldEndsParentFlowInSuccess() throws Exception {
-        // await for the subflow execution
-        CountDownLatch latch = new CountDownLatch(1);
-        AtomicReference<Execution> lastExecution = new AtomicReference<>();
-        Flux<Execution> receivedExecutions = TestsUtils.receive(executionQueue, either -> {
-            Execution exec = either.getLeft();
-            if ("failed-first".equals(exec.getFlowId()) && exec.getState().getCurrent() == State.Type.FAILED) {
-                lastExecution.set(exec);
-                latch.countDown();
-            }
-        });
-
+    public void changeStateInSubflowShouldEndsParentFlowInSuccess(String tenantId) throws Exception {
         // run the parent flow
-        Execution execution = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests", "subflow-parent-of-failed");
+        Execution execution = runnerUtils.runOne(tenantId, NAMESPACE, "subflow-parent-of-failed");
         assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.FAILED);
         assertThat(execution.getTaskRunList()).hasSize(1);
         assertThat(execution.getTaskRunList().getFirst().getState().getCurrent()).isEqualTo(State.Type.FAILED);
 
         // assert on the subflow
-        assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
-        receivedExecutions.blockLast();
-        assertThat(lastExecution.get().getState().getCurrent()).isEqualTo(State.Type.FAILED);
-        assertThat(lastExecution.get().getTaskRunList()).hasSize(1);
-        assertThat(lastExecution.get().getTaskRunList().getFirst().getState().getCurrent()).isEqualTo(State.Type.FAILED);
-
-        // await for the parent execution
-        CountDownLatch parentLatch = new CountDownLatch(1);
-        AtomicReference<Execution> lastParentExecution = new AtomicReference<>();
-        receivedExecutions = TestsUtils.receive(executionQueue, either -> {
-            Execution exec = either.getLeft();
-            if (execution.getId().equals(exec.getId()) && exec.getState().isTerminated()) {
-                lastParentExecution.set(exec);
-                parentLatch.countDown();
-            }
-        });
+        Execution lastExecution = runnerUtils.awaitFlowExecution(e -> e.getState().getCurrent().equals(Type.FAILED), tenantId, NAMESPACE, "failed-first");
+        assertThat(lastExecution.getState().getCurrent()).isEqualTo(State.Type.FAILED);
+        assertThat(lastExecution.getTaskRunList()).hasSize(1);
+        assertThat(lastExecution.getTaskRunList().getFirst().getState().getCurrent()).isEqualTo(State.Type.FAILED);
 
         // restart the subflow
-        Flow flow = flowRepository.findByExecution(lastExecution.get());
-        Execution markedAs = executionService.markAs(lastExecution.get(), flow, lastExecution.get().getTaskRunList().getFirst().getId(), State.Type.SUCCESS);
-        executionQueue.emit(markedAs);
+        Flow flow = flowRepository.findByExecution(lastExecution);
+        Execution markedAs = executionService.markAs(lastExecution, flow, lastExecution.getTaskRunList().getFirst().getId(), State.Type.SUCCESS);
+        runnerUtils.emitAndAwaitExecution(e -> e.getState().isTerminated(), markedAs);
+
+        //We wait for the subflow execution to pass from failed to success
+        Execution lastParentExecution = runnerUtils.awaitFlowExecution(e ->
+            e.getTaskRunList().getFirst().getState().getCurrent().equals(Type.SUCCESS), tenantId, NAMESPACE, "subflow-parent-of-failed");
 
         // assert for the parent flow
-        assertThat(parentLatch.await(10, TimeUnit.SECONDS)).isTrue();
-        receivedExecutions.blockLast();
-        assertThat(lastParentExecution.get().getState().getCurrent()).isEqualTo(State.Type.FAILED); // FIXME should be success but it's FAILED on unit tests
-        assertThat(lastParentExecution.get().getTaskRunList()).hasSize(1);
-        assertThat(lastParentExecution.get().getTaskRunList().getFirst().getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+        assertThat(lastParentExecution.getState().getCurrent()).isEqualTo(State.Type.FAILED); // FIXME should be success but it's FAILED on unit tests
+        assertThat(lastParentExecution.getTaskRunList()).hasSize(1);
+        assertThat(lastParentExecution.getTaskRunList().getFirst().getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
     }
 }

core/src/test/java/io/kestra/core/runners/EmptyVariablesTest.java

@@ -18,7 +18,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 public class EmptyVariablesTest {
 
     @Inject
-    private RunnerUtils runnerUtils;
+    private TestRunnerUtils runnerUtils;
     @Inject
     private FlowInputOutput flowIO;
 

core/src/test/java/io/kestra/core/runners/ExecutionServiceTest.java

@@ -58,7 +58,7 @@ class ExecutionServiceTest {
     LogRepositoryInterface logRepository;
 
     @Inject
-    RunnerUtils runnerUtils;
+    TestRunnerUtils runnerUtils;
 
     @Test
     @LoadFlows({"flows/valids/restart_last_failed.yaml"})

core/src/test/java/io/kestra/core/runners/FlowConcurrencyCaseTest.java

@@ -1,13 +1,15 @@
 package io.kestra.core.runners;
 
 import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.executions.ExecutionKilled;
+import io.kestra.core.models.executions.ExecutionKilledExecution;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.State;
+import io.kestra.core.models.flows.State.History;
 import io.kestra.core.models.flows.State.Type;
 import io.kestra.core.queues.QueueException;
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
-import io.kestra.core.reporter.model.Count;
 import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.services.ExecutionService;
 import io.kestra.core.storages.StorageInterface;
@@ -16,7 +18,6 @@ import jakarta.inject.Inject;
 import jakarta.inject.Named;
 import jakarta.inject.Singleton;
 import org.apache.commons.lang3.StringUtils;
-import reactor.core.publisher.Flux;
 
 import java.io.File;
 import java.io.FileInputStream;
@@ -26,24 +27,21 @@ import java.net.URISyntaxException;
 import java.nio.file.Files;
 import java.time.Duration;
 import java.util.*;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.IntStream;
 
 import static io.kestra.core.tenant.TenantService.MAIN_TENANT;
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertTrue;
 
 @Singleton
 public class FlowConcurrencyCaseTest {
+
+    public static final String NAMESPACE = "io.kestra.tests";
     @Inject
     private StorageInterface storageInterface;
 
     @Inject
-    protected RunnerUtils runnerUtils;
+    protected TestRunnerUtils runnerUtils;
 
     @Inject
     private FlowInputOutput flowIO;
@@ -51,400 +49,192 @@ public class FlowConcurrencyCaseTest {
     @Inject
     private FlowRepositoryInterface flowRepository;
 
-    @Inject
-    @Named(QueueFactoryInterface.EXECUTION_NAMED)
-    protected QueueInterface<Execution> executionQueue;
-
     @Inject
     private ExecutionService executionService;
 
-    public void flowConcurrencyCancel() throws TimeoutException, QueueException, InterruptedException {
-        Execution execution1 = runnerUtils.runOneUntilRunning(MAIN_TENANT, "io.kestra.tests", "flow-concurrency-cancel", null, null, Duration.ofSeconds(30));
-        Execution execution2 = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests", "flow-concurrency-cancel");
+    @Inject
+    @Named(QueueFactoryInterface.KILL_NAMED)
+    protected QueueInterface<ExecutionKilled> killQueue;
+
+    public void flowConcurrencyCancel() throws TimeoutException, QueueException {
+        Execution execution1 = runnerUtils.runOneUntilRunning(MAIN_TENANT, NAMESPACE, "flow-concurrency-cancel", null, null, Duration.ofSeconds(30));
+        Execution execution2 = runnerUtils.runOne(MAIN_TENANT, NAMESPACE, "flow-concurrency-cancel");
 
         assertThat(execution1.getState().isRunning()).isTrue();
         assertThat(execution2.getState().getCurrent()).isEqualTo(State.Type.CANCELLED);
 
-        CountDownLatch latch1 = new CountDownLatch(1);
-
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, e -> {
-            if (e.getLeft().getId().equals(execution1.getId())) {
-                if (e.getLeft().getState().getCurrent() == State.Type.SUCCESS) {
-                    latch1.countDown();
-                }
-            }
-
-            // FIXME we should fail if we receive the cancel execution again but on Kafka it happens
-        });
-
-        assertTrue(latch1.await(1, TimeUnit.MINUTES));
-        receive.blockLast();
+        runnerUtils.awaitExecution(e -> e.getState().getCurrent().equals(Type.SUCCESS), execution1);
     }
 
-    public void flowConcurrencyFail() throws TimeoutException, QueueException, InterruptedException {
-        Execution execution1 = runnerUtils.runOneUntilRunning(MAIN_TENANT, "io.kestra.tests", "flow-concurrency-fail", null, null, Duration.ofSeconds(30));
-        Execution execution2 = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests", "flow-concurrency-fail");
+    public void flowConcurrencyFail() throws TimeoutException, QueueException {
+        Execution execution1 = runnerUtils.runOneUntilRunning(MAIN_TENANT, NAMESPACE, "flow-concurrency-fail", null, null, Duration.ofSeconds(30));
+        Execution execution2 = runnerUtils.runOne(MAIN_TENANT, NAMESPACE, "flow-concurrency-fail");
 
         assertThat(execution1.getState().isRunning()).isTrue();
         assertThat(execution2.getState().getCurrent()).isEqualTo(State.Type.FAILED);
 
-        CountDownLatch latch1 = new CountDownLatch(1);
-
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, e -> {
-            if (e.getLeft().getId().equals(execution1.getId())) {
-                if (e.getLeft().getState().getCurrent() == State.Type.SUCCESS) {
-                    latch1.countDown();
-                }
-            }
-
-            // FIXME we should fail if we receive the cancel execution again but on Kafka it happens
-        });
-
-        assertTrue(latch1.await(1, TimeUnit.MINUTES));
-        receive.blockLast();
+        runnerUtils.awaitExecution(e -> e.getState().getCurrent().equals(Type.SUCCESS), execution1);
     }
 
-    public void flowConcurrencyQueue() throws TimeoutException, QueueException, InterruptedException {
-        Execution execution1 = runnerUtils.runOneUntilRunning(MAIN_TENANT, "io.kestra.tests", "flow-concurrency-queue", null, null, Duration.ofSeconds(30));
+    public void flowConcurrencyQueue() throws QueueException {
+        Execution execution1 = runnerUtils.runOneUntilRunning(MAIN_TENANT, NAMESPACE, "flow-concurrency-queue", null, null, Duration.ofSeconds(30));
         Flow flow = flowRepository
-            .findById(MAIN_TENANT, "io.kestra.tests", "flow-concurrency-queue", Optional.empty())
+            .findById(MAIN_TENANT, NAMESPACE, "flow-concurrency-queue", Optional.empty())
             .orElseThrow();
         Execution execution2 = Execution.newExecution(flow, null, null, Optional.empty());
-        executionQueue.emit(execution2);
+        Execution executionResult2 = runnerUtils.emitAndAwaitExecution(e -> e.getState().getCurrent().equals(Type.SUCCESS), execution2);
+        Execution executionResult1 = runnerUtils.awaitExecution(e -> e.getState().getCurrent().equals(Type.SUCCESS), execution1);
 
         assertThat(execution1.getState().isRunning()).isTrue();
         assertThat(execution2.getState().getCurrent()).isEqualTo(State.Type.CREATED);
 
-        var executionResult1  = new AtomicReference<Execution>();
-        var executionResult2  = new AtomicReference<Execution>();
-
-        CountDownLatch latch1 = new CountDownLatch(1);
-        CountDownLatch latch2 = new CountDownLatch(1);
-        CountDownLatch latch3 = new CountDownLatch(1);
-
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, e -> {
-            if (e.getLeft().getId().equals(execution1.getId())) {
-                executionResult1.set(e.getLeft());
-                if (e.getLeft().getState().getCurrent() == State.Type.SUCCESS) {
-                    latch1.countDown();
-                }
-            }
-
-            if (e.getLeft().getId().equals(execution2.getId())) {
-                executionResult2.set(e.getLeft());
-                if (e.getLeft().getState().getCurrent() == State.Type.RUNNING) {
-                    latch2.countDown();
-                }
-                if (e.getLeft().getState().getCurrent() == State.Type.SUCCESS) {
-                    latch3.countDown();
-                }
-            }
-        });
-
-        assertTrue(latch1.await(1, TimeUnit.MINUTES));
-        assertTrue(latch2.await(1, TimeUnit.MINUTES));
-        assertTrue(latch3.await(1, TimeUnit.MINUTES));
-        receive.blockLast();
-
-        assertThat(executionResult1.get().getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
-        assertThat(executionResult2.get().getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
-        assertThat(executionResult2.get().getState().getHistories().getFirst().getState()).isEqualTo(State.Type.CREATED);
-        assertThat(executionResult2.get().getState().getHistories().get(1).getState()).isEqualTo(State.Type.QUEUED);
-        assertThat(executionResult2.get().getState().getHistories().get(2).getState()).isEqualTo(State.Type.RUNNING);
+        assertThat(executionResult1.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+        assertThat(executionResult2.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+        assertThat(executionResult2.getState().getHistories().getFirst().getState()).isEqualTo(State.Type.CREATED);
+        assertThat(executionResult2.getState().getHistories().get(1).getState()).isEqualTo(State.Type.QUEUED);
+        assertThat(executionResult2.getState().getHistories().get(2).getState()).isEqualTo(State.Type.RUNNING);
     }
 
-    public void flowConcurrencyQueuePause() throws TimeoutException, QueueException, InterruptedException {
-        AtomicReference<String> firstExecutionId = new AtomicReference<>();
-        var firstExecutionResult  = new AtomicReference<Execution>();
-        var secondExecutionResult  = new AtomicReference<Execution>();
-
-        CountDownLatch firstExecutionLatch = new CountDownLatch(1);
-        CountDownLatch secondExecutionLatch = new CountDownLatch(1);
-
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, e -> {
-            if (!"flow-concurrency-queue-pause".equals(e.getLeft().getFlowId())){
-                return;
-            }
-            String currentId = e.getLeft().getId();
-            Type currentState = e.getLeft().getState().getCurrent();
-            if (firstExecutionId.get() == null) {
-                firstExecutionId.set(currentId);
-            }
-
-            if (currentId.equals(firstExecutionId.get())) {
-                if (currentState == State.Type.SUCCESS) {
-                    firstExecutionResult.set(e.getLeft());
-                    firstExecutionLatch.countDown();
-                }
-            } else {
-                if (currentState == State.Type.SUCCESS) {
-                    secondExecutionResult.set(e.getLeft());
-                    secondExecutionLatch.countDown();
-                }
-            }
-        });
-
-
-        Execution execution1 = runnerUtils.runOneUntilPaused(MAIN_TENANT, "io.kestra.tests", "flow-concurrency-queue-pause");
+    public void flowConcurrencyQueuePause() throws QueueException {
+        Execution execution1 = runnerUtils.runOneUntilPaused(MAIN_TENANT, NAMESPACE, "flow-concurrency-queue-pause");
         Flow flow = flowRepository
-            .findById(MAIN_TENANT, "io.kestra.tests", "flow-concurrency-queue-pause", Optional.empty())
+            .findById(MAIN_TENANT, NAMESPACE, "flow-concurrency-queue-pause", Optional.empty())
             .orElseThrow();
         Execution execution2 = Execution.newExecution(flow, null, null, Optional.empty());
-        executionQueue.emit(execution2);
+        Execution secondExecutionResult = runnerUtils.emitAndAwaitExecution(e -> e.getState().getCurrent().equals(Type.SUCCESS), execution2);
+        Execution firstExecutionResult = runnerUtils.awaitExecution(e -> e.getState().getCurrent().equals(Type.SUCCESS), execution1);
 
-        assertThat(execution1.getState().isPaused()).isTrue();
-        assertThat(execution2.getState().getCurrent()).isEqualTo(State.Type.CREATED);
 
-        assertTrue(firstExecutionLatch.await(10, TimeUnit.SECONDS));
-        assertTrue(secondExecutionLatch.await(10, TimeUnit.SECONDS));
-        receive.blockLast();
-
-        assertThat(firstExecutionResult.get().getId()).isEqualTo(execution1.getId());
-        assertThat(firstExecutionResult.get().getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
-        assertThat(secondExecutionResult.get().getId()).isEqualTo(execution2.getId());
-        assertThat(secondExecutionResult.get().getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
-        assertThat(secondExecutionResult.get().getState().getHistories().getFirst().getState()).isEqualTo(State.Type.CREATED);
-        assertThat(secondExecutionResult.get().getState().getHistories().get(1).getState()).isEqualTo(State.Type.QUEUED);
-        assertThat(secondExecutionResult.get().getState().getHistories().get(2).getState()).isEqualTo(State.Type.RUNNING);
+        assertThat(firstExecutionResult.getId()).isEqualTo(execution1.getId());
+        assertThat(firstExecutionResult.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+        assertThat(secondExecutionResult.getId()).isEqualTo(execution2.getId());
+        assertThat(secondExecutionResult.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+        assertThat(secondExecutionResult.getState().getHistories().getFirst().getState()).isEqualTo(State.Type.CREATED);
+        assertThat(secondExecutionResult.getState().getHistories().get(1).getState()).isEqualTo(State.Type.QUEUED);
+        assertThat(secondExecutionResult.getState().getHistories().get(2).getState()).isEqualTo(State.Type.RUNNING);
     }
 
-    public void flowConcurrencyCancelPause() throws TimeoutException, QueueException, InterruptedException {
-        AtomicReference<String> firstExecutionId = new AtomicReference<>();
-        var firstExecutionResult  = new AtomicReference<Execution>();
-        var secondExecutionResult  = new AtomicReference<Execution>();
-        CountDownLatch firstExecLatch = new CountDownLatch(1);
-        CountDownLatch secondExecLatch = new CountDownLatch(1);
-
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, e -> {
-            if (!"flow-concurrency-cancel-pause".equals(e.getLeft().getFlowId())){
-                return;
-            }
-            String currentId = e.getLeft().getId();
-            Type currentState = e.getLeft().getState().getCurrent();
-            if (firstExecutionId.get() == null) {
-                firstExecutionId.set(currentId);
-            }
-            if (currentId.equals(firstExecutionId.get())) {
-                if (currentState == State.Type.SUCCESS) {
-                    firstExecutionResult.set(e.getLeft());
-                    firstExecLatch.countDown();
-                }
-            } else {
-                if (currentState == State.Type.CANCELLED) {
-                    secondExecutionResult.set(e.getLeft());
-                    secondExecLatch.countDown();
-                }
-            }
-        });
-
-        Execution execution1 = runnerUtils.runOneUntilPaused(MAIN_TENANT, "io.kestra.tests", "flow-concurrency-cancel-pause");
+    public void flowConcurrencyCancelPause() throws QueueException {
+        Execution execution1 = runnerUtils.runOneUntilPaused(MAIN_TENANT, NAMESPACE, "flow-concurrency-cancel-pause");
         Flow flow = flowRepository
-            .findById(MAIN_TENANT, "io.kestra.tests", "flow-concurrency-cancel-pause", Optional.empty())
+            .findById(MAIN_TENANT, NAMESPACE, "flow-concurrency-cancel-pause", Optional.empty())
             .orElseThrow();
         Execution execution2 = Execution.newExecution(flow, null, null, Optional.empty());
-        executionQueue.emit(execution2);
+        Execution secondExecutionResult = runnerUtils.emitAndAwaitExecution(e -> e.getState().getCurrent().equals(Type.CANCELLED), execution2);
+        Execution firstExecutionResult = runnerUtils.awaitExecution(e -> e.getState().getCurrent().equals(Type.SUCCESS), execution1);
 
-        assertThat(execution1.getState().isPaused()).isTrue();
-        assertThat(execution2.getState().getCurrent()).isEqualTo(State.Type.CREATED);
 
-        assertTrue(firstExecLatch.await(10, TimeUnit.SECONDS));
-        assertTrue(secondExecLatch.await(10, TimeUnit.SECONDS));
-        receive.blockLast();
-
-        assertThat(firstExecutionResult.get().getId()).isEqualTo(execution1.getId());
-        assertThat(firstExecutionResult.get().getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
-        assertThat(secondExecutionResult.get().getId()).isEqualTo(execution2.getId());
-        assertThat(secondExecutionResult.get().getState().getCurrent()).isEqualTo(State.Type.CANCELLED);
-        assertThat(secondExecutionResult.get().getState().getHistories().getFirst().getState()).isEqualTo(State.Type.CREATED);
-        assertThat(secondExecutionResult.get().getState().getHistories().get(1).getState()).isEqualTo(State.Type.CANCELLED);
+        assertThat(firstExecutionResult.getId()).isEqualTo(execution1.getId());
+        assertThat(firstExecutionResult.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+        assertThat(secondExecutionResult.getId()).isEqualTo(execution2.getId());
+        assertThat(secondExecutionResult.getState().getCurrent()).isEqualTo(State.Type.CANCELLED);
+        assertThat(secondExecutionResult.getState().getHistories().getFirst().getState()).isEqualTo(State.Type.CREATED);
+        assertThat(secondExecutionResult.getState().getHistories().get(1).getState()).isEqualTo(State.Type.CANCELLED);
     }
 
-    public void flowConcurrencyWithForEachItem() throws TimeoutException, QueueException, InterruptedException, URISyntaxException, IOException {
-        URI file = storageUpload();
+    public void flowConcurrencyWithForEachItem(String tenantId) throws QueueException, URISyntaxException, IOException {
+        URI file = storageUpload(tenantId);
         Map<String, Object> inputs = Map.of("file", file.toString(), "batch", 4);
-        Execution forEachItem = runnerUtils.runOneUntilRunning(MAIN_TENANT, "io.kestra.tests", "flow-concurrency-for-each-item", null,
+        Execution forEachItem = runnerUtils.runOneUntilRunning(tenantId, NAMESPACE, "flow-concurrency-for-each-item", null,
         (flow, execution1) -> flowIO.readExecutionInputs(flow, execution1, inputs), Duration.ofSeconds(5));
         assertThat(forEachItem.getState().getCurrent()).isEqualTo(Type.RUNNING);
 
-        Set<String> executionIds = new HashSet<>();
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, e -> {
-            if ("flow-concurrency-queue".equals(e.getLeft().getFlowId()) && e.getLeft().getState().isRunning()) {
-                executionIds.add(e.getLeft().getId());
-            }
-        });
 
-        // wait a little to be sure there are not too many executions started
-        Thread.sleep(500);
-
-        assertThat(executionIds).hasSize(1);
-        receive.blockLast();
-
-        Execution terminated = runnerUtils.awaitExecution(e -> e.getId().equals(forEachItem.getId()) && e.getState().isTerminated(), () -> {}, Duration.ofSeconds(10));
+        Execution terminated = runnerUtils.awaitExecution(e -> e.getState().isTerminated(),forEachItem);
         assertThat(terminated.getState().getCurrent()).isEqualTo(Type.SUCCESS);
+
+        List<Execution> executions = runnerUtils.awaitFlowExecutionNumber(2, tenantId, NAMESPACE, "flow-concurrency-queue");
+
+        assertThat(executions).extracting(e -> e.getState().getCurrent()).containsOnly(Type.SUCCESS);
+        assertThat(executions.stream()
+            .map(e -> e.getState().getHistories())
+            .flatMap(List::stream)
+            .map(History::getState)
+            .toList()).contains(Type.QUEUED);
     }
 
     public void flowConcurrencyQueueRestarted() throws Exception {
-        Execution execution1 = runnerUtils.runOneUntilRunning(MAIN_TENANT, "io.kestra.tests", "flow-concurrency-queue-fail", null, null, Duration.ofSeconds(30));
+        Execution execution1 = runnerUtils.runOneUntilRunning(MAIN_TENANT, NAMESPACE,
+            "flow-concurrency-queue-fail", null, null, Duration.ofSeconds(30));
         Flow flow = flowRepository
-            .findById(MAIN_TENANT, "io.kestra.tests", "flow-concurrency-queue-fail", Optional.empty())
+            .findById(MAIN_TENANT, NAMESPACE, "flow-concurrency-queue-fail", Optional.empty())
             .orElseThrow();
         Execution execution2 = Execution.newExecution(flow, null, null, Optional.empty());
-        executionQueue.emit(execution2);
+        runnerUtils.emitAndAwaitExecution(e -> e.getState().getCurrent().equals(Type.RUNNING), execution2);
 
-        assertThat(execution1.getState().isRunning()).isTrue();
-        assertThat(execution2.getState().getCurrent()).isEqualTo(State.Type.CREATED);
-
-        var executionResult1  = new AtomicReference<Execution>();
-        var executionResult2  = new AtomicReference<Execution>();
-
-        CountDownLatch latch1 = new CountDownLatch(2);
-        AtomicReference<Execution> failedExecution = new AtomicReference<>();
-        CountDownLatch latch2 = new CountDownLatch(1);
-        CountDownLatch latch3 = new CountDownLatch(1);
-
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, e -> {
-            if (e.getLeft().getId().equals(execution1.getId())) {
-                executionResult1.set(e.getLeft());
-                if (e.getLeft().getState().getCurrent() == Type.FAILED) {
-                    failedExecution.set(e.getLeft());
-                    latch1.countDown();
-                }
-            }
-
-            if (e.getLeft().getId().equals(execution2.getId())) {
-                executionResult2.set(e.getLeft());
-                if (e.getLeft().getState().getCurrent() == State.Type.RUNNING) {
-                    latch2.countDown();
-                }
-                if (e.getLeft().getState().getCurrent() == Type.FAILED) {
-                    latch3.countDown();
-                }
-            }
-        });
-
-        assertTrue(latch2.await(1, TimeUnit.MINUTES));
-        assertThat(failedExecution.get()).isNotNull();
         // here the first fail and the second is now running.
         // we restart the first one, it should be queued then fail again.
-        Execution restarted = executionService.restart(failedExecution.get(), null);
-        executionQueue.emit(restarted);
+        Execution failedExecution = runnerUtils.awaitExecution(e -> e.getState().getCurrent().equals(Type.FAILED), execution1);
+        Execution restarted = executionService.restart(failedExecution, null);
+        Execution executionResult1 = runnerUtils.restartExecution(e -> e.getState().getCurrent().equals(Type.FAILED), restarted);
+        Execution executionResult2 = runnerUtils.awaitExecution(e -> e.getState().getCurrent().equals(Type.FAILED), execution2);
 
-        assertTrue(latch3.await(1, TimeUnit.MINUTES));
-        assertTrue(latch1.await(1, TimeUnit.MINUTES));
-        receive.blockLast();
-
-        assertThat(executionResult1.get().getState().getCurrent()).isEqualTo(Type.FAILED);
+        assertThat(executionResult1.getState().getCurrent()).isEqualTo(Type.FAILED);
         // it should have been queued after restarted
-        assertThat(executionResult1.get().getState().getHistories().stream().anyMatch(history -> history.getState() == Type.RESTARTED)).isTrue();
-        assertThat(executionResult1.get().getState().getHistories().stream().anyMatch(history -> history.getState() == Type.QUEUED)).isTrue();
-        assertThat(executionResult2.get().getState().getCurrent()).isEqualTo(Type.FAILED);
-        assertThat(executionResult2.get().getState().getHistories().getFirst().getState()).isEqualTo(State.Type.CREATED);
-        assertThat(executionResult2.get().getState().getHistories().get(1).getState()).isEqualTo(State.Type.QUEUED);
-        assertThat(executionResult2.get().getState().getHistories().get(2).getState()).isEqualTo(State.Type.RUNNING);
+        assertThat(executionResult1.getState().getHistories().stream().anyMatch(history -> history.getState() == Type.RESTARTED)).isTrue();
+        assertThat(executionResult1.getState().getHistories().stream().anyMatch(history -> history.getState() == Type.QUEUED)).isTrue();
+        assertThat(executionResult2.getState().getCurrent()).isEqualTo(Type.FAILED);
+        assertThat(executionResult2.getState().getHistories().getFirst().getState()).isEqualTo(State.Type.CREATED);
+        assertThat(executionResult2.getState().getHistories().get(1).getState()).isEqualTo(State.Type.QUEUED);
+        assertThat(executionResult2.getState().getHistories().get(2).getState()).isEqualTo(State.Type.RUNNING);
     }
 
-    public void flowConcurrencyQueueAfterExecution() throws TimeoutException, QueueException, InterruptedException {
-        Execution execution1 = runnerUtils.runOneUntilRunning(MAIN_TENANT, "io.kestra.tests", "flow-concurrency-queue-after-execution", null, null, Duration.ofSeconds(30));
+    public void flowConcurrencyQueueAfterExecution() throws QueueException {
+        Execution execution1 = runnerUtils.runOneUntilRunning(MAIN_TENANT, NAMESPACE, "flow-concurrency-queue-after-execution", null, null, Duration.ofSeconds(30));
         Flow flow = flowRepository
-            .findById(MAIN_TENANT, "io.kestra.tests", "flow-concurrency-queue-after-execution", Optional.empty())
+            .findById(MAIN_TENANT, NAMESPACE, "flow-concurrency-queue-after-execution", Optional.empty())
             .orElseThrow();
         Execution execution2 = Execution.newExecution(flow, null, null, Optional.empty());
-        executionQueue.emit(execution2);
+        Execution executionResult2 = runnerUtils.emitAndAwaitExecution(e -> e.getState().getCurrent().equals(Type.SUCCESS), execution2);
+        Execution executionResult1 = runnerUtils.awaitExecution(e -> e.getState().getCurrent().equals(Type.SUCCESS), execution1);
 
-        assertThat(execution1.getState().isRunning()).isTrue();
-        assertThat(execution2.getState().getCurrent()).isEqualTo(State.Type.CREATED);
-
-        var executionResult1  = new AtomicReference<Execution>();
-        var executionResult2  = new AtomicReference<Execution>();
-
-        CountDownLatch latch1 = new CountDownLatch(1);
-        CountDownLatch latch2 = new CountDownLatch(1);
-        CountDownLatch latch3 = new CountDownLatch(1);
-
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, e -> {
-            if (e.getLeft().getId().equals(execution1.getId())) {
-                executionResult1.set(e.getLeft());
-                if (e.getLeft().getState().getCurrent() == State.Type.SUCCESS) {
-                    latch1.countDown();
-                }
-            }
-
-            if (e.getLeft().getId().equals(execution2.getId())) {
-                executionResult2.set(e.getLeft());
-                if (e.getLeft().getState().getCurrent() == State.Type.RUNNING) {
-                    latch2.countDown();
-                }
-                if (e.getLeft().getState().getCurrent() == State.Type.SUCCESS) {
-                    latch3.countDown();
-                }
-            }
-        });
-
-        assertTrue(latch1.await(1, TimeUnit.MINUTES));
-        assertTrue(latch2.await(1, TimeUnit.MINUTES));
-        assertTrue(latch3.await(1, TimeUnit.MINUTES));
-        receive.blockLast();
-
-        assertThat(executionResult1.get().getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
-        assertThat(executionResult2.get().getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
-        assertThat(executionResult2.get().getState().getHistories().getFirst().getState()).isEqualTo(State.Type.CREATED);
-        assertThat(executionResult2.get().getState().getHistories().get(1).getState()).isEqualTo(State.Type.QUEUED);
-        assertThat(executionResult2.get().getState().getHistories().get(2).getState()).isEqualTo(State.Type.RUNNING);
+        assertThat(executionResult1.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+        assertThat(executionResult2.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+        assertThat(executionResult2.getState().getHistories().getFirst().getState()).isEqualTo(State.Type.CREATED);
+        assertThat(executionResult2.getState().getHistories().get(1).getState()).isEqualTo(State.Type.QUEUED);
+        assertThat(executionResult2.getState().getHistories().get(2).getState()).isEqualTo(State.Type.RUNNING);
     }
 
-    public void flowConcurrencySubflow() throws TimeoutException, QueueException, InterruptedException {
-        CountDownLatch successLatch = new CountDownLatch(1);
-        CountDownLatch canceledLatch = new CountDownLatch(1);
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, e -> {
-            if (e.getLeft().getFlowId().equals("flow-concurrency-cancel")) {
-                if (e.getLeft().getState().getCurrent() == State.Type.SUCCESS) {
-                    successLatch.countDown();
-                }
-                if (e.getLeft().getState().getCurrent() == Type.CANCELLED) {
-                    canceledLatch.countDown();
-                }
-            }
+    public void flowConcurrencySubflow(String tenantId) throws TimeoutException, QueueException {
+        runnerUtils.runOneUntilRunning(tenantId, NAMESPACE, "flow-concurrency-subflow", null, null, Duration.ofSeconds(30));
+        runnerUtils.runOne(tenantId, NAMESPACE, "flow-concurrency-subflow");
 
-            // FIXME we should fail if we receive the cancel execution again but on Kafka it happens
-        });
-
-        Execution execution1 = runnerUtils.runOneUntilRunning(MAIN_TENANT, "io.kestra.tests", "flow-concurrency-subflow", null, null, Duration.ofSeconds(30));
-        Execution execution2 = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests", "flow-concurrency-subflow");
-
-        assertThat(execution1.getState().isRunning()).isTrue();
-        assertThat(execution2.getState().getCurrent()).isEqualTo(Type.SUCCESS);
-
-        // assert we have one canceled subflow and one in success
-        assertTrue(canceledLatch.await(1, TimeUnit.MINUTES));
-        assertTrue(successLatch.await(1, TimeUnit.MINUTES));
-        receive.blockLast();
+        List<Execution> subFlowExecs = runnerUtils.awaitFlowExecutionNumber(2, tenantId, NAMESPACE, "flow-concurrency-cancel");
+        assertThat(subFlowExecs).extracting(e -> e.getState().getCurrent()).containsExactlyInAnyOrder(Type.SUCCESS, Type.CANCELLED);
 
         // run another execution to be sure that everything work (purge is correctly done)
-        CountDownLatch newSuccessLatch = new CountDownLatch(1);
-        Flux<Execution> secondReceive = TestsUtils.receive(executionQueue, e -> {
-            if (e.getLeft().getFlowId().equals("flow-concurrency-cancel")) {
-                if (e.getLeft().getState().getCurrent() == State.Type.SUCCESS) {
-                    newSuccessLatch.countDown();
-                }
-            }
-
-            // FIXME we should fail if we receive the cancel execution again but on Kafka it happens
-        });
-        Execution execution3 = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests", "flow-concurrency-subflow");
+        Execution execution3 = runnerUtils.runOne(tenantId, NAMESPACE, "flow-concurrency-subflow");
         assertThat(execution3.getState().getCurrent()).isEqualTo(Type.SUCCESS);
-
-        // assert we have two successful subflow
-        assertTrue(newSuccessLatch.await(1, TimeUnit.MINUTES));
-        secondReceive.blockLast();
+        runnerUtils.awaitFlowExecution(e -> e.getState().getCurrent().equals(Type.SUCCESS), tenantId, NAMESPACE, "flow-concurrency-cancel");
     }
 
-    private URI storageUpload() throws URISyntaxException, IOException {
+    public void flowConcurrencyParallelSubflowKill() throws QueueException {
+        Execution parent = runnerUtils.runOneUntilRunning(MAIN_TENANT, NAMESPACE, "flow-concurrency-parallel-subflow-kill", null, null, Duration.ofSeconds(30));
+        Execution queued = runnerUtils.awaitFlowExecution(e -> e.getState().isQueued(), MAIN_TENANT, NAMESPACE, "flow-concurrency-parallel-subflow-kill-child");
+
+        // Kill the parent
+        killQueue.emit(ExecutionKilledExecution
+            .builder()
+            .state(ExecutionKilled.State.REQUESTED)
+            .executionId(parent.getId())
+            .isOnKillCascade(true)
+            .tenantId(MAIN_TENANT)
+            .build()
+        );
+
+        Execution terminated = runnerUtils.awaitExecution(e -> e.getState().isTerminated(), queued);
+        assertThat(terminated.getState().getCurrent()).isEqualTo(State.Type.KILLED);
+        assertThat(terminated.getState().getHistories().stream().noneMatch(h -> h.getState() == Type.RUNNING)).isTrue();
+        assertThat(terminated.getTaskRunList()).isNull();
+    }
+
+    private URI storageUpload(String tenantId) throws URISyntaxException, IOException {
         File tempFile = File.createTempFile("file", ".txt");
 
         Files.write(tempFile.toPath(), content());
 
         return storageInterface.put(
-            MAIN_TENANT,
+            tenantId,
             null,
             new URI("/file/storage/file.txt"),
             new FileInputStream(tempFile)

core/src/test/java/io/kestra/core/runners/FlowTriggerCaseTest.java

@@ -2,82 +2,61 @@ package io.kestra.core.runners;
 
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.State;
+import io.kestra.core.models.flows.State.Type;
 import io.kestra.core.queues.QueueException;
-import io.kestra.core.queues.QueueFactoryInterface;
-import io.kestra.core.queues.QueueInterface;
-import io.kestra.core.utils.TestsUtils;
 import jakarta.inject.Inject;
-import jakarta.inject.Named;
 import jakarta.inject.Singleton;
-import reactor.core.publisher.Flux;
 
-import java.util.ArrayList;
+import java.time.Instant;
+import java.util.Comparator;
 import java.util.List;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicReference;
 
 import static io.kestra.core.tenant.TenantService.MAIN_TENANT;
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertTrue;
 
 @Singleton
 public class FlowTriggerCaseTest {
-    @Inject
-    @Named(QueueFactoryInterface.EXECUTION_NAMED)
-    protected QueueInterface<Execution> executionQueue;
+
+    public static final String NAMESPACE = "io.kestra.tests.trigger";
 
     @Inject
-    protected RunnerUtils runnerUtils;
+    protected TestRunnerUtils runnerUtils;
 
-    public void trigger() throws InterruptedException, TimeoutException, QueueException {
-        CountDownLatch countDownLatch = new CountDownLatch(3);
-        AtomicReference<Execution> flowListener = new AtomicReference<>();
-        AtomicReference<Execution> flowListenerNoInput = new AtomicReference<>();
-        AtomicReference<Execution> flowListenerNamespace = new AtomicReference<>();
-
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, either -> {
-            Execution execution = either.getLeft();
-            if (execution.getState().getCurrent() == State.Type.SUCCESS) {
-                if (flowListenerNoInput.get() == null && execution.getFlowId().equals("trigger-flow-listener-no-inputs")) {
-                    flowListenerNoInput.set(execution);
-                    countDownLatch.countDown();
-                } else if (flowListener.get() == null && execution.getFlowId().equals("trigger-flow-listener")) {
-                    flowListener.set(execution);
-                    countDownLatch.countDown();
-                } else if (flowListenerNamespace.get() == null && execution.getFlowId().equals("trigger-flow-listener-namespace-condition")) {
-                    flowListenerNamespace.set(execution);
-                    countDownLatch.countDown();
-                }
-            }
-        });
-
-        Execution execution = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests.trigger", "trigger-flow");
+    public void trigger(String tenantId) throws InterruptedException, TimeoutException, QueueException {
+        Execution execution = runnerUtils.runOne(tenantId, NAMESPACE, "trigger-flow");
 
         assertThat(execution.getTaskRunList().size()).isEqualTo(1);
         assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
 
-        assertTrue(countDownLatch.await(15, TimeUnit.SECONDS), "remaining countdown: %s".formatted(countDownLatch.getCount()));
-        receive.blockLast();
+        Execution flowListenerNoInput = runnerUtils.awaitFlowExecution(
+            e -> e.getState().getCurrent().equals(Type.SUCCESS), tenantId, NAMESPACE,
+            "trigger-flow-listener-no-inputs");
+        Execution flowListener = runnerUtils.awaitFlowExecution(
+            e -> e.getState().getCurrent().equals(Type.SUCCESS), tenantId, NAMESPACE,
+            "trigger-flow-listener");
+        Execution flowListenerNamespace = runnerUtils.awaitFlowExecution(
+            e -> e.getState().getCurrent().equals(Type.SUCCESS), tenantId, NAMESPACE,
+            "trigger-flow-listener-namespace-condition");
 
-        assertThat(flowListener.get().getTaskRunList().size()).isEqualTo(1);
-        assertThat(flowListener.get().getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
-        assertThat(flowListener.get().getTaskRunList().getFirst().getOutputs().get("value")).isEqualTo("childs: from parents: " + execution.getId());
-        assertThat(flowListener.get().getTrigger().getVariables().get("executionId")).isEqualTo(execution.getId());
-        assertThat(flowListener.get().getTrigger().getVariables().get("namespace")).isEqualTo("io.kestra.tests.trigger");
-        assertThat(flowListener.get().getTrigger().getVariables().get("flowId")).isEqualTo("trigger-flow");
 
-        assertThat(flowListenerNoInput.get().getTaskRunList().size()).isEqualTo(1);
-        assertThat(flowListenerNoInput.get().getTrigger().getVariables().get("executionId")).isEqualTo(execution.getId());
-        assertThat(flowListenerNoInput.get().getTrigger().getVariables().get("namespace")).isEqualTo("io.kestra.tests.trigger");
-        assertThat(flowListenerNoInput.get().getTrigger().getVariables().get("flowId")).isEqualTo("trigger-flow");
-        assertThat(flowListenerNoInput.get().getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+        assertThat(flowListener.getTaskRunList().size()).isEqualTo(1);
+        assertThat(flowListener.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+        assertThat(flowListener.getTaskRunList().getFirst().getOutputs().get("value")).isEqualTo("childs: from parents: " + execution.getId());
+        assertThat(flowListener.getTrigger().getVariables().get("executionId")).isEqualTo(execution.getId());
+        assertThat(flowListener.getTrigger().getVariables().get("namespace")).isEqualTo(NAMESPACE);
+        assertThat(flowListener.getTrigger().getVariables().get("flowId")).isEqualTo("trigger-flow");
 
-        assertThat(flowListenerNamespace.get().getTaskRunList().size()).isEqualTo(1);
-        assertThat(flowListenerNamespace.get().getTrigger().getVariables().get("namespace")).isEqualTo("io.kestra.tests.trigger");
+        assertThat(flowListenerNoInput.getTaskRunList().size()).isEqualTo(1);
+        assertThat(flowListenerNoInput.getTrigger().getVariables().get("executionId")).isEqualTo(execution.getId());
+        assertThat(flowListenerNoInput.getTrigger().getVariables().get("namespace")).isEqualTo(NAMESPACE);
+        assertThat(flowListenerNoInput.getTrigger().getVariables().get("flowId")).isEqualTo("trigger-flow");
+        assertThat(flowListenerNoInput.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+
+        assertThat(flowListenerNamespace.getTaskRunList().size()).isEqualTo(1);
+        assertThat(flowListenerNamespace.getTrigger().getVariables().get("namespace")).isEqualTo(NAMESPACE);
         // it will be triggered for 'trigger-flow' or any of the 'trigger-flow-listener*', so we only assert that it's one of them
-        assertThat(flowListenerNamespace.get().getTrigger().getVariables().get("flowId"))
+        assertThat(flowListenerNamespace.getTrigger().getVariables().get("flowId"))
             .satisfiesAnyOf(
                 arg -> assertThat(arg).isEqualTo("trigger-flow"),
                 arg -> assertThat(arg).isEqualTo("trigger-flow-listener-no-inputs"),
@@ -85,56 +64,43 @@ public class FlowTriggerCaseTest {
             );
     }
 
-    public void triggerWithPause() throws InterruptedException, TimeoutException, QueueException {
-        CountDownLatch countDownLatch = new CountDownLatch(4);
-        List<Execution> flowListeners = new ArrayList<>();
-
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, either -> {
-            Execution execution = either.getLeft();
-            if (execution.getState().getCurrent() == State.Type.SUCCESS && execution.getFlowId().equals("trigger-flow-listener-with-pause")) {
-                flowListeners.add(execution);
-                countDownLatch.countDown();
-            }
-        });
-
+    public void triggerWithPause() throws TimeoutException, QueueException {
         Execution execution = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests.trigger.pause", "trigger-flow-with-pause");
 
         assertThat(execution.getTaskRunList().size()).isEqualTo(3);
         assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
 
-        assertTrue(countDownLatch.await(15, TimeUnit.SECONDS));
-        receive.blockLast();
+        List<Execution> triggeredExec = runnerUtils.awaitFlowExecutionNumber(
+            4,
+            MAIN_TENANT,
+            "io.kestra.tests.trigger.pause",
+            "trigger-flow-listener-with-pause");
 
-        assertThat(flowListeners.size()).isEqualTo(4);
-        assertThat(flowListeners.get(0).getOutputs().get("status")).isEqualTo("RUNNING");
-        assertThat(flowListeners.get(1).getOutputs().get("status")).isEqualTo("PAUSED");
-        assertThat(flowListeners.get(2).getOutputs().get("status")).isEqualTo("RUNNING");
-        assertThat(flowListeners.get(3).getOutputs().get("status")).isEqualTo("SUCCESS");
+        assertThat(triggeredExec.size()).isEqualTo(4);
+        List<Execution> sortedExecs = triggeredExec.stream()
+            .sorted(Comparator.comparing(e -> e.getState().getEndDate().orElse(Instant.now())))
+            .toList();
+        assertThat(sortedExecs.get(0).getOutputs().get("status")).isEqualTo("RUNNING");
+        assertThat(sortedExecs.get(1).getOutputs().get("status")).isEqualTo("PAUSED");
+        assertThat(sortedExecs.get(2).getOutputs().get("status")).isEqualTo("RUNNING");
+        assertThat(sortedExecs.get(3).getOutputs().get("status")).isEqualTo("SUCCESS");
     }
 
-    public void triggerWithConcurrencyLimit() throws QueueException, TimeoutException, InterruptedException {
-        CountDownLatch countDownLatch = new CountDownLatch(5);
-        List<Execution> flowListeners = new ArrayList<>();
+    public void triggerWithConcurrencyLimit(String tenantId) throws QueueException, TimeoutException {
+        Execution execution1 = runnerUtils.runOneUntilRunning(tenantId, "io.kestra.tests.trigger.concurrency", "trigger-flow-with-concurrency-limit");
+        Execution execution2 = runnerUtils.runOne(tenantId, "io.kestra.tests.trigger.concurrency", "trigger-flow-with-concurrency-limit");
 
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, either -> {
-            Execution execution = either.getLeft();
-            if (execution.getState().getCurrent() == State.Type.SUCCESS && execution.getFlowId().equals("trigger-flow-listener-with-concurrency-limit")) {
-                flowListeners.add(execution);
-                countDownLatch.countDown();
-            }
-        });
+        List<Execution> triggeredExec = runnerUtils.awaitFlowExecutionNumber(
+            5,
+            tenantId,
+            "io.kestra.tests.trigger.concurrency",
+            "trigger-flow-listener-with-concurrency-limit");
 
-        Execution execution1 = runnerUtils.runOneUntilRunning(MAIN_TENANT, "io.kestra.tests.trigger.concurrency", "trigger-flow-with-concurrency-limit");
-        Execution execution2 = runnerUtils.runOneUntilRunning(MAIN_TENANT, "io.kestra.tests.trigger.concurrency", "trigger-flow-with-concurrency-limit");
-
-        assertTrue(countDownLatch.await(15, TimeUnit.SECONDS));
-        receive.blockLast();
-
-        assertThat(flowListeners.size()).isEqualTo(5);
-        assertThat(flowListeners.stream().anyMatch(e -> e.getOutputs().get("status").equals("RUNNING") && e.getOutputs().get("executionId").equals(execution1.getId()))).isTrue();
-        assertThat(flowListeners.stream().anyMatch(e -> e.getOutputs().get("status").equals("SUCCESS") && e.getOutputs().get("executionId").equals(execution1.getId()))).isTrue();
-        assertThat(flowListeners.stream().anyMatch(e -> e.getOutputs().get("status").equals("QUEUED") && e.getOutputs().get("executionId").equals(execution2.getId()))).isTrue();
-        assertThat(flowListeners.stream().anyMatch(e -> e.getOutputs().get("status").equals("RUNNING") && e.getOutputs().get("executionId").equals(execution2.getId()))).isTrue();
-        assertThat(flowListeners.stream().anyMatch(e -> e.getOutputs().get("status").equals("SUCCESS") && e.getOutputs().get("executionId").equals(execution2.getId()))).isTrue();
+        assertThat(triggeredExec.size()).isEqualTo(5);
+        assertThat(triggeredExec.stream().anyMatch(e -> e.getOutputs().get("status").equals("RUNNING") && e.getOutputs().get("executionId").equals(execution1.getId()))).isTrue();
+        assertThat(triggeredExec.stream().anyMatch(e -> e.getOutputs().get("status").equals("SUCCESS") && e.getOutputs().get("executionId").equals(execution1.getId()))).isTrue();
+        assertThat(triggeredExec.stream().anyMatch(e -> e.getOutputs().get("status").equals("QUEUED") && e.getOutputs().get("executionId").equals(execution2.getId()))).isTrue();
+        assertThat(triggeredExec.stream().anyMatch(e -> e.getOutputs().get("status").equals("RUNNING") && e.getOutputs().get("executionId").equals(execution2.getId()))).isTrue();
+        assertThat(triggeredExec.stream().anyMatch(e -> e.getOutputs().get("status").equals("SUCCESS") && e.getOutputs().get("executionId").equals(execution2.getId()))).isTrue();
     }
 }

core/src/test/java/io/kestra/core/runners/InputsTest.java

@@ -48,7 +48,7 @@ public class InputsTest {
     private QueueInterface<LogEntry> logQueue;
 
     @Inject
-    private RunnerUtils runnerUtils;
+    private TestRunnerUtils runnerUtils;
 
     public static Map<String, Object> inputs = ImmutableMap.<String, Object>builder()
         .put("string", "myString")

core/src/test/java/io/kestra/core/runners/ListenersTest.java

@@ -24,7 +24,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 class ListenersTest {
 
     @Inject
-    private RunnerUtils runnerUtils;
+    private TestRunnerUtils runnerUtils;
 
     @Inject
     private LocalFlowRepositoryLoader repositoryLoader;

core/src/test/java/io/kestra/core/runners/MultipleConditionTriggerCaseTest.java

@@ -1,244 +1,193 @@
 package io.kestra.core.runners;
 
+import io.kestra.core.models.flows.State.Type;
 import io.kestra.core.queues.QueueException;
-import io.kestra.core.utils.TestsUtils;
+import io.kestra.core.repositories.ArrayListTotal;
+import io.kestra.core.repositories.ExecutionRepositoryInterface;
 import io.micronaut.context.ApplicationContext;
 import io.kestra.core.models.executions.Execution;
-import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.State;
-import io.kestra.core.queues.QueueFactoryInterface;
-import io.kestra.core.queues.QueueInterface;
 import io.kestra.core.repositories.FlowRepositoryInterface;
 
-import java.time.Duration;
+import io.micronaut.data.model.Pageable;
+
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicReference;
 
 import jakarta.inject.Inject;
-import jakarta.inject.Named;
 import jakarta.inject.Singleton;
-import reactor.core.publisher.Flux;
 
 import static io.kestra.core.tenant.TenantService.MAIN_TENANT;
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertTrue;
 
 @Singleton
 public class MultipleConditionTriggerCaseTest {
 
-    @Inject
-    @Named(QueueFactoryInterface.EXECUTION_NAMED)
-    protected QueueInterface<Execution> executionQueue;
+    public static final String NAMESPACE = "io.kestra.tests.trigger";
 
     @Inject
-    protected RunnerUtils runnerUtils;
+    protected TestRunnerUtils runnerUtils;
 
     @Inject
     protected FlowRepositoryInterface flowRepository;
 
+    @Inject
+    protected ExecutionRepositoryInterface executionRepository;
+
     @Inject
     protected ApplicationContext applicationContext;
 
     public void trigger() throws InterruptedException, TimeoutException, QueueException {
-        CountDownLatch countDownLatch = new CountDownLatch(3);
-        ConcurrentHashMap<String, Execution> ended = new ConcurrentHashMap<>();
-        List<String> watchedExecutions = List.of("trigger-multiplecondition-flow-a",
-            "trigger-multiplecondition-flow-b",
-            "trigger-multiplecondition-listener"
-        );
-
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, either -> {
-            Execution execution = either.getLeft();
-            if (watchedExecutions.contains(execution.getFlowId()) && execution.getState().getCurrent() == State.Type.SUCCESS) {
-                ended.put(execution.getId(), execution);
-                countDownLatch.countDown();
-            }
-        });
-
         // first one
-        Execution execution = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests.trigger",
-            "trigger-multiplecondition-flow-a", Duration.ofSeconds(60));
+        Execution execution = runnerUtils.runOne(MAIN_TENANT, NAMESPACE, "trigger-multiplecondition-flow-a");
         assertThat(execution.getTaskRunList().size()).isEqualTo(1);
         assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
 
         // wait a little to be sure that the trigger is not launching execution
         Thread.sleep(1000);
-        assertThat(ended.size()).isEqualTo(1);
+        ArrayListTotal<Execution> flowBExecutions = executionRepository.findByFlowId(MAIN_TENANT,
+            NAMESPACE, "trigger-multiplecondition-flow-b", Pageable.UNPAGED);
+        ArrayListTotal<Execution> listenerExecutions = executionRepository.findByFlowId(MAIN_TENANT,
+            NAMESPACE, "trigger-multiplecondition-listener", Pageable.UNPAGED);
+        assertThat(flowBExecutions).isEmpty();
+        assertThat(listenerExecutions).isEmpty();
 
         // second one
-        execution = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests.trigger",
-            "trigger-multiplecondition-flow-b", Duration.ofSeconds(60));
+        execution = runnerUtils.runOne(MAIN_TENANT, NAMESPACE, "trigger-multiplecondition-flow-b");
         assertThat(execution.getTaskRunList().size()).isEqualTo(1);
         assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
 
         // trigger is done
-        assertTrue(countDownLatch.await(10, TimeUnit.SECONDS));
-        receive.blockLast();
-        assertThat(ended.size()).isEqualTo(3);
-
-        Flow flow = flowRepository.findById(MAIN_TENANT, "io.kestra.tests.trigger",
-            "trigger-multiplecondition-listener").orElseThrow();
-        Execution triggerExecution = ended.entrySet()
-            .stream()
-            .filter(e -> e.getValue().getFlowId().equals(flow.getId()))
-            .findFirst()
-            .map(Map.Entry::getValue)
-            .orElseThrow();
+        Execution triggerExecution = runnerUtils.awaitFlowExecution(
+            e -> e.getState().getCurrent().equals(Type.SUCCESS),
+            MAIN_TENANT, NAMESPACE, "trigger-multiplecondition-listener");
 
         assertThat(triggerExecution.getTaskRunList().size()).isEqualTo(1);
         assertThat(triggerExecution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
 
         assertThat(triggerExecution.getTrigger().getVariables().get("executionId")).isEqualTo(execution.getId());
-        assertThat(triggerExecution.getTrigger().getVariables().get("namespace")).isEqualTo("io.kestra.tests.trigger");
+        assertThat(triggerExecution.getTrigger().getVariables().get("namespace")).isEqualTo(
+            NAMESPACE);
         assertThat(triggerExecution.getTrigger().getVariables().get("flowId")).isEqualTo("trigger-multiplecondition-flow-b");
     }
 
-    public void failed() throws InterruptedException, TimeoutException, QueueException {
-        CountDownLatch countDownLatch = new CountDownLatch(1);
-        AtomicReference<Execution> listener = new AtomicReference<>();
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, either -> {
-            Execution execution = either.getLeft();
-            if (execution.getFlowId().equals("trigger-flow-listener-namespace-condition")
-                && execution.getState().getCurrent().isTerminated()) {
-                listener.set(execution);
-                countDownLatch.countDown();
-            }
-        });
-
+    public void failed(String tenantId) throws InterruptedException, TimeoutException, QueueException {
         // first one
-        Execution execution = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests.trigger",
-            "trigger-multiplecondition-flow-c", Duration.ofSeconds(60));
+        Execution execution = runnerUtils.runOne(tenantId, NAMESPACE,
+            "trigger-multiplecondition-flow-c");
         assertThat(execution.getTaskRunList().size()).isEqualTo(1);
         assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.FAILED);
 
         // wait a little to be sure that the trigger is not launching execution
         Thread.sleep(1000);
-        assertThat(listener.get()).isNull();
+        ArrayListTotal<Execution> byFlowId = executionRepository.findByFlowId(tenantId, NAMESPACE,
+            "trigger-multiplecondition-flow-d", Pageable.UNPAGED);
+        assertThat(byFlowId).isEmpty();
 
         // second one
-        execution = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests.trigger",
-            "trigger-multiplecondition-flow-d", Duration.ofSeconds(60));
+        execution = runnerUtils.runOne(tenantId, NAMESPACE,
+            "trigger-multiplecondition-flow-d");
         assertThat(execution.getTaskRunList().size()).isEqualTo(1);
         assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
 
+        Execution triggerExecution = runnerUtils.awaitFlowExecution(
+            e -> e.getState().getCurrent().equals(Type.SUCCESS),
+            tenantId, NAMESPACE, "trigger-flow-listener-namespace-condition");
+
         // trigger was not done
-        assertTrue(countDownLatch.await(10, TimeUnit.SECONDS));
-        receive.blockLast();
-        assertThat(listener.get()).isNotNull();
-        assertThat(listener.get().getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+        assertThat(triggerExecution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
     }
 
-    public void flowTriggerPreconditions()
-        throws InterruptedException, TimeoutException, QueueException {
-        CountDownLatch countDownLatch = new CountDownLatch(1);
-        AtomicReference<Execution> flowTrigger = new AtomicReference<>();
-
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, either -> {
-            Execution execution = either.getLeft();
-            if (execution.getState().getCurrent() == State.Type.SUCCESS && execution.getFlowId()
-                .equals("flow-trigger-preconditions-flow-listen")) {
-                flowTrigger.set(execution);
-                countDownLatch.countDown();
-            }
-        });
+    public void flowTriggerPreconditions() throws TimeoutException, QueueException {
 
         // flowA
         Execution execution = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests.trigger.preconditions",
-            "flow-trigger-preconditions-flow-a", Duration.ofSeconds(60));
+            "flow-trigger-preconditions-flow-a");
         assertThat(execution.getTaskRunList().size()).isEqualTo(1);
         assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
 
         // flowB: we trigger it two times, as flow-trigger-flow-preconditions-flow-listen is configured with resetOnSuccess: false it should be triggered two times
         execution = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests.trigger.preconditions",
-            "flow-trigger-preconditions-flow-a", Duration.ofSeconds(60));
+            "flow-trigger-preconditions-flow-a");
         assertThat(execution.getTaskRunList().size()).isEqualTo(1);
         assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
         execution = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests.trigger.preconditions",
-            "flow-trigger-preconditions-flow-b", Duration.ofSeconds(60));
+            "flow-trigger-preconditions-flow-b");
         assertThat(execution.getTaskRunList().size()).isEqualTo(1);
         assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
 
         // trigger is done
-        assertTrue(countDownLatch.await(1, TimeUnit.SECONDS));
-        receive.blockLast();
-        assertThat(flowTrigger.get()).isNotNull();
+        Execution triggerExecution = runnerUtils.awaitFlowExecution(
+            e -> e.getState().getCurrent().equals(Type.SUCCESS),
+            MAIN_TENANT, "io.kestra.tests.trigger.preconditions", "flow-trigger-preconditions-flow-listen");
 
-        Execution triggerExecution = flowTrigger.get();
         assertThat(triggerExecution.getTaskRunList().size()).isEqualTo(1);
         assertThat(triggerExecution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
         assertThat(triggerExecution.getTrigger().getVariables().get("outputs")).isNotNull();
         assertThat((Map<String, Object>) triggerExecution.getTrigger().getVariables().get("outputs")).containsEntry("some", "value");
     }
 
-    public void flowTriggerPreconditionsMergeOutputs() throws QueueException, TimeoutException, InterruptedException {
+    public void flowTriggerPreconditionsMergeOutputs(String tenantId) throws QueueException, TimeoutException {
         // we do the same as in flowTriggerPreconditions() but we trigger flows in the opposite order to be sure that outputs are merged
-        CountDownLatch countDownLatch = new CountDownLatch(1);
-        AtomicReference<Execution> flowTrigger = new AtomicReference<>();
-
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, either -> {
-            Execution execution = either.getLeft();
-            if (execution.getState().getCurrent() == State.Type.SUCCESS && execution.getFlowId()
-                .equals("flow-trigger-preconditions-flow-listen")) {
-                flowTrigger.set(execution);
-                countDownLatch.countDown();
-            }
-        });
 
         // flowB
-        Execution execution = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests.trigger.preconditions",
-            "flow-trigger-preconditions-flow-b", Duration.ofSeconds(60));
+        Execution execution = runnerUtils.runOne(tenantId, "io.kestra.tests.trigger.preconditions",
+            "flow-trigger-preconditions-flow-b");
         assertThat(execution.getTaskRunList().size()).isEqualTo(1);
         assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
 
         // flowA
-        execution = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests.trigger.preconditions",
-            "flow-trigger-preconditions-flow-a", Duration.ofSeconds(60));
+        execution = runnerUtils.runOne(tenantId, "io.kestra.tests.trigger.preconditions",
+            "flow-trigger-preconditions-flow-a");
         assertThat(execution.getTaskRunList().size()).isEqualTo(1);
         assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
 
         // trigger is done
-        assertTrue(countDownLatch.await(1, TimeUnit.SECONDS));
-        receive.blockLast();
-        assertThat(flowTrigger.get()).isNotNull();
+        Execution triggerExecution = runnerUtils.awaitFlowExecution(
+            e -> e.getState().getCurrent().equals(Type.SUCCESS),
+            tenantId, "io.kestra.tests.trigger.preconditions", "flow-trigger-preconditions-flow-listen");
 
-        Execution triggerExecution = flowTrigger.get();
         assertThat(triggerExecution.getTaskRunList().size()).isEqualTo(1);
         assertThat(triggerExecution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
         assertThat(triggerExecution.getTrigger().getVariables().get("outputs")).isNotNull();
         assertThat((Map<String, Object>) triggerExecution.getTrigger().getVariables().get("outputs")).containsEntry("some", "value");
     }
 
-    public void flowTriggerOnPaused()
-        throws InterruptedException, TimeoutException, QueueException {
-        CountDownLatch countDownLatch = new CountDownLatch(1);
-        AtomicReference<Execution> flowTrigger = new AtomicReference<>();
-
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, either -> {
-            Execution execution = either.getLeft();
-            if (execution.getState().getCurrent() == State.Type.SUCCESS && execution.getFlowId()
-                .equals("flow-trigger-paused-listen")) {
-                flowTrigger.set(execution);
-                countDownLatch.countDown();
-            }
-        });
-
+    public void flowTriggerOnPaused() throws TimeoutException, QueueException {
         Execution execution = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests.trigger.paused",
-            "flow-trigger-paused-flow", Duration.ofSeconds(60));
+            "flow-trigger-paused-flow");
         assertThat(execution.getTaskRunList().size()).isEqualTo(2);
         assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
 
         // trigger is done
-        assertTrue(countDownLatch.await(1, TimeUnit.SECONDS));
-        receive.blockLast();
-        assertThat(flowTrigger.get()).isNotNull();
+        Execution triggerExecution = runnerUtils.awaitFlowExecution(
+            e -> e.getState().getCurrent().equals(Type.SUCCESS),
+            MAIN_TENANT, "io.kestra.tests.trigger.paused", "flow-trigger-paused-listen");
 
-        Execution triggerExecution = flowTrigger.get();
         assertThat(triggerExecution.getTaskRunList().size()).isEqualTo(1);
         assertThat(triggerExecution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
     }
+
+    public void forEachItemWithFlowTrigger() throws TimeoutException, QueueException {
+        Execution execution = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests.trigger.foreachitem",
+            "flow-trigger-for-each-item-parent");
+        assertThat(execution.getTaskRunList().size()).isEqualTo(5);
+        assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+
+        // trigger is done
+        List<Execution> childExecutions = runnerUtils.awaitFlowExecutionNumber(5, MAIN_TENANT, "io.kestra.tests.trigger.foreachitem", "flow-trigger-for-each-item-child");
+        assertThat(childExecutions).hasSize(5);
+        childExecutions.forEach(exec -> {
+            assertThat(exec.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+            assertThat(exec.getTaskRunList().size()).isEqualTo(1);
+        });
+
+        List<Execution> grandchildExecutions = runnerUtils.awaitFlowExecutionNumber(5, MAIN_TENANT, "io.kestra.tests.trigger.foreachitem", "flow-trigger-for-each-item-grandchild");
+        assertThat(grandchildExecutions).hasSize(5);
+        grandchildExecutions.forEach(exec -> {
+            assertThat(exec.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+            assertThat(exec.getTaskRunList().size()).isEqualTo(2);
+        });
+    }
 }

core/src/test/java/io/kestra/core/runners/PluginDefaultsCaseTest.java

@@ -30,7 +30,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 @Singleton
 public class PluginDefaultsCaseTest {
     @Inject
-    private RunnerUtils runnerUtils;
+    private TestRunnerUtils runnerUtils;
 
     public void taskDefaults() throws TimeoutException, QueueException {
         Execution execution = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests", "plugin-defaults", Duration.ofSeconds(60));

core/src/test/java/io/kestra/core/runners/RestartCaseTest.java

@@ -4,29 +4,19 @@ 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.State;
-import io.kestra.core.queues.QueueException;
-import io.kestra.core.queues.QueueFactoryInterface;
-import io.kestra.core.queues.QueueInterface;
+import io.kestra.core.models.flows.State.Type;
 import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.services.ExecutionService;
 
 import java.time.Duration;
 import java.util.List;
 import java.util.Optional;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicReference;
 
-import io.kestra.core.utils.TestsUtils;
 import jakarta.inject.Inject;
-import jakarta.inject.Named;
 import jakarta.inject.Singleton;
-import reactor.core.publisher.Flux;
 
 import static io.kestra.core.tenant.TenantService.MAIN_TENANT;
 import static org.assertj.core.api.Assertions.assertThat;
-import static io.kestra.core.utils.Rethrow.throwRunnable;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 @Singleton
@@ -35,38 +25,30 @@ public class RestartCaseTest {
     private FlowRepositoryInterface flowRepository;
 
     @Inject
-    private RunnerUtils runnerUtils;
+    private TestRunnerUtils runnerUtils;
 
     @Inject
     private ExecutionService executionService;
 
-    @Inject
-    @Named(QueueFactoryInterface.EXECUTION_NAMED)
-    private QueueInterface<Execution> executionQueue;
-
     public void restartFailedThenSuccess() throws Exception {
         Flow flow = flowRepository.findById(MAIN_TENANT, "io.kestra.tests", "restart_last_failed").orElseThrow();
 
-        Execution firstExecution = runnerUtils.runOne(MAIN_TENANT, flow.getNamespace(), flow.getId(), Duration.ofSeconds(60));
+        Execution firstExecution = runnerUtils.runOne(MAIN_TENANT, flow.getNamespace(), flow.getId());
 
         assertThat(firstExecution.getState().getCurrent()).isEqualTo(State.Type.FAILED);
         assertThat(firstExecution.getTaskRunList()).hasSize(3);
         assertThat(firstExecution.getTaskRunList().get(2).getState().getCurrent()).isEqualTo(State.Type.FAILED);
 
         // wait
-        Execution finishedRestartedExecution = runnerUtils.awaitExecution(
+        Execution restartedExec = executionService.restart(firstExecution, null);
+        assertThat(restartedExec).isNotNull();
+        assertThat(restartedExec.getId()).isEqualTo(firstExecution.getId());
+        assertThat(restartedExec.getParentId()).isNull();
+        assertThat(restartedExec.getTaskRunList().size()).isEqualTo(3);
+        assertThat(restartedExec.getState().getCurrent()).isEqualTo(State.Type.RESTARTED);
+        Execution finishedRestartedExecution = runnerUtils.restartExecution(
             execution -> execution.getState().getCurrent() == State.Type.SUCCESS && execution.getId().equals(firstExecution.getId()),
-            throwRunnable(() -> {
-                Execution restartedExec = executionService.restart(firstExecution, null);
-                assertThat(restartedExec).isNotNull();
-                assertThat(restartedExec.getId()).isEqualTo(firstExecution.getId());
-                assertThat(restartedExec.getParentId()).isNull();
-                assertThat(restartedExec.getTaskRunList().size()).isEqualTo(3);
-                assertThat(restartedExec.getState().getCurrent()).isEqualTo(State.Type.RESTARTED);
-
-                executionQueue.emit(restartedExec);
-            }),
-            Duration.ofSeconds(60)
+            restartedExec
         );
 
         assertThat(finishedRestartedExecution).isNotNull();
@@ -93,19 +75,16 @@ public class RestartCaseTest {
         assertThat(firstExecution.getTaskRunList().getFirst().getState().getCurrent()).isEqualTo(State.Type.FAILED);
 
         // wait
-        Execution finishedRestartedExecution = runnerUtils.awaitExecution(
-            execution -> execution.getState().getCurrent() == State.Type.FAILED && execution.getTaskRunList().getFirst().getAttempts().size() == 2,
-            throwRunnable(() -> {
-                Execution restartedExec = executionService.restart(firstExecution, null);
-                executionQueue.emit(restartedExec);
+        Execution restartedExec = executionService.restart(firstExecution, null);
 
-                assertThat(restartedExec).isNotNull();
-                assertThat(restartedExec.getId()).isEqualTo(firstExecution.getId());
-                assertThat(restartedExec.getParentId()).isNull();
-                assertThat(restartedExec.getTaskRunList().size()).isEqualTo(1);
-                assertThat(restartedExec.getState().getCurrent()).isEqualTo(State.Type.RESTARTED);
-            }),
-            Duration.ofSeconds(60)
+        assertThat(restartedExec).isNotNull();
+        assertThat(restartedExec.getId()).isEqualTo(firstExecution.getId());
+        assertThat(restartedExec.getParentId()).isNull();
+        assertThat(restartedExec.getTaskRunList().size()).isEqualTo(1);
+        assertThat(restartedExec.getState().getCurrent()).isEqualTo(State.Type.RESTARTED);
+        Execution finishedRestartedExecution = runnerUtils.restartExecution(
+            execution -> execution.getState().getCurrent() == State.Type.FAILED && execution.getTaskRunList().getFirst().getAttempts().size() == 2,
+            restartedExec
         );
 
         assertThat(finishedRestartedExecution).isNotNull();
@@ -128,19 +107,16 @@ public class RestartCaseTest {
         assertThat(firstExecution.getTaskRunList().get(3).getState().getCurrent()).isEqualTo(State.Type.FAILED);
 
         // wait
-        Execution finishedRestartedExecution = runnerUtils.awaitExecution(
-            execution -> execution.getState().getCurrent() == State.Type.FAILED && execution.findTaskRunsByTaskId("failStep").stream().findFirst().get().getAttempts().size() == 2,
-            throwRunnable(() -> {
-                Execution restartedExec = executionService.restart(firstExecution, null);
-                executionQueue.emit(restartedExec);
+        Execution restartedExec = executionService.restart(firstExecution, null);
 
-                assertThat(restartedExec).isNotNull();
-                assertThat(restartedExec.getId()).isEqualTo(firstExecution.getId());
-                assertThat(restartedExec.getParentId()).isNull();
-                assertThat(restartedExec.getTaskRunList().size()).isEqualTo(4);
-                assertThat(restartedExec.getState().getCurrent()).isEqualTo(State.Type.RESTARTED);
-            }),
-            Duration.ofSeconds(60)
+        assertThat(restartedExec).isNotNull();
+        assertThat(restartedExec.getId()).isEqualTo(firstExecution.getId());
+        assertThat(restartedExec.getParentId()).isNull();
+        assertThat(restartedExec.getTaskRunList().size()).isEqualTo(4);
+        assertThat(restartedExec.getState().getCurrent()).isEqualTo(State.Type.RESTARTED);
+        Execution finishedRestartedExecution = runnerUtils.restartExecution(
+            execution -> execution.getState().getCurrent() == State.Type.FAILED && execution.findTaskRunsByTaskId("failStep").stream().findFirst().get().getAttempts().size() == 2,
+            restartedExec
         );
 
         assertThat(finishedRestartedExecution).isNotNull();
@@ -163,21 +139,19 @@ public class RestartCaseTest {
         assertThat(firstExecution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
 
         // wait
+        Execution restartedExec = executionService.replay(firstExecution, firstExecution.findTaskRunByTaskIdAndValue("2_end", List.of()).getId(), null);
+
+        assertThat(restartedExec.getState().getCurrent()).isEqualTo(State.Type.RESTARTED);
+        assertThat(restartedExec.getState().getHistories()).hasSize(4);
+        assertThat(restartedExec.getTaskRunList()).hasSize(20);
+        assertThat(restartedExec.getTaskRunList().get(19).getState().getCurrent()).isEqualTo(State.Type.RESTARTED);
+
+        assertThat(restartedExec.getId()).isNotEqualTo(firstExecution.getId());
+        assertThat(restartedExec.getTaskRunList().get(1).getId()).isNotEqualTo(firstExecution.getTaskRunList().get(1).getId());
         Execution finishedRestartedExecution = runnerUtils.awaitChildExecution(
             flow,
             firstExecution,
-            throwRunnable(() -> {
-                Execution restartedExec = executionService.replay(firstExecution, firstExecution.findTaskRunByTaskIdAndValue("2_end", List.of()).getId(), null);
-                executionQueue.emit(restartedExec);
-
-                assertThat(restartedExec.getState().getCurrent()).isEqualTo(State.Type.RESTARTED);
-                assertThat(restartedExec.getState().getHistories()).hasSize(4);
-                assertThat(restartedExec.getTaskRunList()).hasSize(20);
-                assertThat(restartedExec.getTaskRunList().get(19).getState().getCurrent()).isEqualTo(State.Type.RESTARTED);
-
-                assertThat(restartedExec.getId()).isNotEqualTo(firstExecution.getId());
-                assertThat(restartedExec.getTaskRunList().get(1).getId()).isNotEqualTo(firstExecution.getTaskRunList().get(1).getId());
-            }),
+            restartedExec,
             Duration.ofSeconds(60)
         );
 
@@ -195,71 +169,53 @@ public class RestartCaseTest {
         Execution restart = executionService.restart(execution, null);
         assertThat(restart.getState().getCurrent()).isEqualTo(State.Type.RESTARTED);
 
-        Execution restartEnded = runnerUtils.awaitExecution(
+        Execution restartEnded = runnerUtils.restartExecution(
             e -> e.getState().getCurrent() == State.Type.FAILED,
-            throwRunnable(() -> executionQueue.emit(restart)),
-            Duration.ofSeconds(120)
+            restart
         );
 
         assertThat(restartEnded.getState().getCurrent()).isEqualTo(State.Type.FAILED);
 
         Execution newRestart = executionService.restart(restartEnded, null);
 
-        restartEnded = runnerUtils.awaitExecution(
+        restartEnded = runnerUtils.restartExecution(
             e -> e.getState().getCurrent() == State.Type.FAILED,
-            throwRunnable(() -> executionQueue.emit(newRestart)),
-            Duration.ofSeconds(120)
+            newRestart
         );
 
         assertThat(restartEnded.getState().getCurrent()).isEqualTo(State.Type.FAILED);
     }
 
     public void restartSubflow() throws Exception {
-        CountDownLatch countDownLatch = new CountDownLatch(1);
-        Flux<Execution> receiveSubflows = TestsUtils.receive(executionQueue, either -> {
-            Execution subflowExecution = either.getLeft();
-            if (subflowExecution.getFlowId().equals("restart-child") && subflowExecution.getState().getCurrent().isFailed()) {
-                countDownLatch.countDown();
-            }
-        });
-
         Execution execution = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests", "restart-parent");
         assertThat(execution.getTaskRunList()).hasSize(3);
         assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.FAILED);
 
         // here we must have 1 failed subflows
-        assertTrue(countDownLatch.await(1, TimeUnit.MINUTES));
-        receiveSubflows.blockLast();
+        runnerUtils.awaitFlowExecution(e -> e.getState().getCurrent().isFailed(), MAIN_TENANT, "io.kestra.tests", "restart-child");
 
         // there is 3 values so we must restart it 3 times to end the 3 subflows
-        CountDownLatch successLatch = new CountDownLatch(3);
-        receiveSubflows = TestsUtils.receive(executionQueue, either -> {
-            Execution subflowExecution = either.getLeft();
-            if (subflowExecution.getFlowId().equals("restart-child") && subflowExecution.getState().getCurrent().isSuccess()) {
-                successLatch.countDown();
-            }
-        });
         Execution restarted1 = executionService.restart(execution, null);
-        execution = runnerUtils.awaitExecution(
+        execution = runnerUtils.restartExecution(
             e -> e.getState().getCurrent() == State.Type.FAILED && e.getFlowId().equals("restart-parent"),
-            throwRunnable(() -> executionQueue.emit(restarted1)),
-            Duration.ofSeconds(10)
+            restarted1
         );
         Execution restarted2 = executionService.restart(execution, null);
-        execution = runnerUtils.awaitExecution(
+        execution = runnerUtils.restartExecution(
             e -> e.getState().getCurrent() == State.Type.FAILED && e.getFlowId().equals("restart-parent"),
-            throwRunnable(() -> executionQueue.emit(restarted2)),
-            Duration.ofSeconds(10)
+            restarted2
         );
         Execution restarted3 = executionService.restart(execution, null);
-        execution = runnerUtils.awaitExecution(
+        execution = runnerUtils.restartExecution(
             e -> e.getState().getCurrent() == State.Type.SUCCESS && e.getFlowId().equals("restart-parent"),
-            throwRunnable(() -> executionQueue.emit(restarted3)),
-            Duration.ofSeconds(10)
+            restarted3
         );
         assertThat(execution.getTaskRunList()).hasSize(6);
-        assertTrue(successLatch.await(1, TimeUnit.MINUTES));
-        receiveSubflows.blockLast();
+
+        List<Execution> childExecutions = runnerUtils.awaitFlowExecutionNumber(3, MAIN_TENANT, "io.kestra.tests", "restart-child");
+        List<Execution> successfulRestart = childExecutions.stream()
+            .filter(e -> e.getState().getCurrent().equals(Type.SUCCESS)).toList();
+        assertThat(successfulRestart).hasSize(3);
     }
 
     public void restartFailedWithFinally() throws Exception {
@@ -272,19 +228,15 @@ public class RestartCaseTest {
         assertThat(firstExecution.getTaskRunList().get(1).getState().getCurrent()).isEqualTo(State.Type.FAILED);
 
         // wait
-        Execution finishedRestartedExecution = runnerUtils.awaitExecution(
-            execution -> executionService.isTerminated(flow, execution) && execution.getState().isSuccess() && execution.getId().equals(firstExecution.getId()),
-            throwRunnable(() -> {
-                Execution restartedExec = executionService.restart(firstExecution, null);
-                assertThat(restartedExec).isNotNull();
-                assertThat(restartedExec.getId()).isEqualTo(firstExecution.getId());
-                assertThat(restartedExec.getParentId()).isNull();
-                assertThat(restartedExec.getTaskRunList().size()).isEqualTo(2);
-                assertThat(restartedExec.getState().getCurrent()).isEqualTo(State.Type.RESTARTED);
-
-                executionQueue.emit(restartedExec);
-            }),
-            Duration.ofSeconds(60)
+        Execution restartedExec = executionService.restart(firstExecution, null);
+        assertThat(restartedExec).isNotNull();
+        assertThat(restartedExec.getId()).isEqualTo(firstExecution.getId());
+        assertThat(restartedExec.getParentId()).isNull();
+        assertThat(restartedExec.getTaskRunList().size()).isEqualTo(2);
+        assertThat(restartedExec.getState().getCurrent()).isEqualTo(State.Type.RESTARTED);
+        Execution finishedRestartedExecution = runnerUtils.restartExecution(
+            execution -> executionService.isTerminated(flow, execution) && execution.getState().isSuccess(),
+            restartedExec
         );
 
         assertThat(finishedRestartedExecution).isNotNull();
@@ -309,21 +261,17 @@ public class RestartCaseTest {
         assertThat(firstExecution.getTaskRunList().get(1).getState().getCurrent()).isEqualTo(State.Type.FAILED);
 
         // wait
-        Execution finishedRestartedExecution = runnerUtils.awaitExecution(
-            execution -> executionService.isTerminated(flow, execution) && execution.getState().isSuccess() && execution.getId().equals(firstExecution.getId()),
-            throwRunnable(() -> {
-                Execution restartedExec = executionService.restart(firstExecution, null);
-                assertThat(restartedExec).isNotNull();
-                assertThat(restartedExec.getId()).isEqualTo(firstExecution.getId());
-                assertThat(restartedExec.getParentId()).isNull();
-                assertThat(restartedExec.getTaskRunList().size()).isEqualTo(2);
-                assertThat(restartedExec.getState().getCurrent()).isEqualTo(State.Type.RESTARTED);
+        Execution restartedExec = executionService.restart(firstExecution, null);
+        assertThat(restartedExec).isNotNull();
+        assertThat(restartedExec.getId()).isEqualTo(firstExecution.getId());
+        assertThat(restartedExec.getParentId()).isNull();
+        assertThat(restartedExec.getTaskRunList().size()).isEqualTo(2);
+        assertThat(restartedExec.getState().getCurrent()).isEqualTo(State.Type.RESTARTED);
 
-                executionQueue.emit(restartedExec);
-            }),
-            Duration.ofSeconds(60)
+        Execution finishedRestartedExecution = runnerUtils.restartExecution(
+            execution -> executionService.isTerminated(flow, execution) && execution.getState().isSuccess(),
+            restartedExec
         );
-
         assertThat(finishedRestartedExecution).isNotNull();
         assertThat(finishedRestartedExecution.getId()).isEqualTo(firstExecution.getId());
         assertThat(finishedRestartedExecution.getParentId()).isNull();

core/src/test/java/io/kestra/core/runners/RunContextTest.java

@@ -98,7 +98,7 @@ class RunContextTest {
     private FlowInputOutput flowIO;
 
     @Inject
-    private RunnerUtils runnerUtils;
+    private TestRunnerUtils runnerUtils;
 
     @Inject
     protected LocalFlowRepositoryLoader repositoryLoader;

core/src/test/java/io/kestra/core/runners/SLATestCase.java

@@ -16,7 +16,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 @Singleton
 public class SLATestCase {
     @Inject
-    private RunnerUtils runnerUtils;
+    private TestRunnerUtils runnerUtils;
 
     public void maxDurationSLAShouldFail() throws QueueException, TimeoutException {
         Execution execution = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests", "sla-max-duration-fail");
@@ -36,14 +36,14 @@ public class SLATestCase {
         assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
     }
 
-    public void executionConditionSLAShouldCancel() throws QueueException, TimeoutException {
-        Execution execution = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests", "sla-execution-condition", null, (f, e) -> Map.of("string", "CANCEL"));
+    public void executionConditionSLAShouldCancel(String tenantId) throws QueueException, TimeoutException {
+        Execution execution = runnerUtils.runOne(tenantId, "io.kestra.tests", "sla-execution-condition", null, (f, e) -> Map.of("string", "CANCEL"));
 
         assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.CANCELLED);
     }
 
-    public void executionConditionSLAShouldLabel() throws QueueException, TimeoutException {
-        Execution execution = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests", "sla-execution-condition", null, (f, e) -> Map.of("string", "LABEL"));
+    public void executionConditionSLAShouldLabel(String tenantId) throws QueueException, TimeoutException {
+        Execution execution = runnerUtils.runOne(tenantId, "io.kestra.tests", "sla-execution-condition", null, (f, e) -> Map.of("string", "LABEL"));
 
         assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
         assertThat(execution.getLabels()).contains(new Label("sla", "violated"));

core/src/test/java/io/kestra/core/runners/ScheduleDateCaseTest.java

@@ -3,54 +3,31 @@ package io.kestra.core.runners;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.State;
+import io.kestra.core.models.flows.State.Type;
 import io.kestra.core.queues.QueueException;
-import io.kestra.core.queues.QueueFactoryInterface;
-import io.kestra.core.queues.QueueInterface;
 import io.kestra.core.repositories.FlowRepositoryInterface;
-import io.kestra.core.utils.TestsUtils;
 import jakarta.inject.Inject;
-import jakarta.inject.Named;
 import jakarta.inject.Singleton;
-import reactor.core.publisher.Flux;
 
 import java.time.ZonedDateTime;
 import java.util.Optional;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
 
-import static io.kestra.core.tenant.TenantService.MAIN_TENANT;
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertTrue;
 
 @Singleton
 public class ScheduleDateCaseTest {
     @Inject
     private FlowRepositoryInterface flowRepository;
-
     @Inject
-    @Named(QueueFactoryInterface.EXECUTION_NAMED)
-    protected QueueInterface<Execution> executionQueue;
+    private TestRunnerUtils runnerUtils;
 
-    public void shouldScheduleOnDate() throws QueueException, InterruptedException {
+    public void shouldScheduleOnDate(String tenantId) throws QueueException {
         ZonedDateTime scheduleOn = ZonedDateTime.now().plusSeconds(1);
-        Flow flow = flowRepository.findById(MAIN_TENANT, "io.kestra.tests", "minimal").orElseThrow();
+        Flow flow = flowRepository.findById(tenantId, "io.kestra.tests", "minimal").orElseThrow();
         Execution execution = Execution.newExecution(flow, null, null, Optional.of(scheduleOn));
-        this.executionQueue.emit(execution);
-
-        assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.CREATED);
         assertThat(execution.getScheduleDate()).isEqualTo(scheduleOn.toInstant());
+        assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.CREATED);
 
-        CountDownLatch latch1 = new CountDownLatch(1);
-
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, e -> {
-            if (e.getLeft().getId().equals(execution.getId())) {
-                if (e.getLeft().getState().getCurrent() == State.Type.SUCCESS) {
-                    latch1.countDown();
-                }
-            }
-        });
-
-        assertTrue(latch1.await(1, TimeUnit.MINUTES));
-        receive.blockLast();
+        runnerUtils.emitAndAwaitExecution(e -> e.getState().getCurrent().equals(Type.SUCCESS), execution);
     }
 }