fix: disabled flakky test shouldGetReport

Add some refactoring to allow some methods to be overrided

.devcontainer/README.md

@@ -23,7 +23,15 @@ In the meantime, you can move onto the next step...
 
 ---
 
+### Requirements
+
+- Java 21 (LTS versions).
+  > ⚠️ Java 24 and above are not supported yet and will fail with `invalid source release: 21`.
+- Gradle (comes with wrapper `./gradlew`)
+- Docker (optional, for running Kestra in containers)
+
 ### Development:
+
 - (Optional) By default, your dev server will target `localhost:8080`. If your backend is running elsewhere, you can create `.env.development.local` under `ui` folder with this content:
 ```
 VITE_APP_API_URL={myApiUrl}

.github/workflows/auto-translate-ui-keys.yml

@@ -26,7 +26,7 @@ jobs:
             fetch-depth: 0
 
         - name: Set up Python
-          uses: actions/setup-python@v5
+          uses: actions/setup-python@v6
           with:
             python-version: "3.x"
 
@@ -39,7 +39,7 @@ jobs:
             OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
 
         - name: Set up Node
-          uses: actions/setup-node@v4
+          uses: actions/setup-node@v5
           with:
             node-version: "20.x"
 

.github/workflows/codeql-analysis.yml

@@ -50,7 +50,7 @@ jobs:
 
     # Set up JDK
     - name: Set up JDK
-      uses: actions/setup-java@v4
+      uses: actions/setup-java@v5
       if: ${{ matrix.language == 'java' }}
       with:
         distribution: 'temurin'

.github/workflows/e2e.yml

@@ -37,7 +37,7 @@ jobs:
           path: kestra
 
       # Setup build
-      - uses: kestra-io/actions/.github/actions/setup-build@main
+      - uses: kestra-io/actions/composite/setup-build@main
         name: Setup - Build
         id: build
         with:

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

@@ -4,7 +4,7 @@ on:
   workflow_dispatch:
     inputs:
       releaseVersion:
-        description: 'The release version (e.g., 0.21.0-rc1)'
+        description: 'The release version (e.g., 0.21.0)'
         required: true
         type: string
       nextVersion:
@@ -25,21 +25,13 @@ jobs:
         with:
           fetch-depth: 0
 
-      # Checkout GitHub Actions
-      - uses: actions/checkout@v5
-        with:
-          repository: kestra-io/actions
-          path: actions
-          ref: main
-
       # Setup build
-      - uses: ./actions/.github/actions/setup-build
+      - uses: kestra-io/actions/composite/setup-build@main
         id: build
         with:
           java-enabled: true
           node-enabled: true
           python-enabled: true
-          caches-enabled: true
 
       # Get Plugins List
       - name: Get Plugins List
@@ -60,7 +52,7 @@ jobs:
           GITHUB_PAT: ${{ secrets.GH_PERSONAL_TOKEN }}
         run: |
           chmod +x ./dev-tools/release-plugins.sh;
-          
+
           ./dev-tools/release-plugins.sh \
           --release-version=${{github.event.inputs.releaseVersion}} \
           --next-version=${{github.event.inputs.nextVersion}} \
@@ -73,10 +65,10 @@ jobs:
           GITHUB_PAT: ${{ secrets.GH_PERSONAL_TOKEN }}
         run: |
           chmod +x ./dev-tools/release-plugins.sh;
-          
+
           ./dev-tools/release-plugins.sh \
           --release-version=${{github.event.inputs.releaseVersion}} \
           --next-version=${{github.event.inputs.nextVersion}} \
           --dry-run \
           --yes \
-          ${{ steps.plugins-list.outputs.repositories }}
+          ${{ steps.plugins-list.outputs.repositories }}

.github/workflows/gradle-release.yml

@@ -4,7 +4,7 @@ on:
   workflow_dispatch:
     inputs:
       releaseVersion:
-        description: 'The release version (e.g., 0.21.0-rc1)'
+        description: 'The release version (e.g., 0.21.0)'
         required: true
         type: string
       nextVersion:
@@ -23,8 +23,8 @@ jobs:
       # Checks
       - name: Check Inputs
         run: |
-          if ! [[ "$RELEASE_VERSION" =~ ^[0-9]+(\.[0-9]+)\.0-rc[01](-SNAPSHOT)?$ ]]; then
-            echo "Invalid release version. Must match regex: ^[0-9]+(\.[0-9]+)\.0-rc[01](-SNAPSHOT)?$"
+          if ! [[ "$RELEASE_VERSION" =~ ^[0-9]+(\.[0-9]+)\.0$ ]]; then
+            echo "Invalid release version. Must match regex: ^[0-9]+(\.[0-9]+)\.0$"
             exit 1
           fi
 
@@ -38,15 +38,8 @@ jobs:
           fetch-depth: 0
           path: kestra
 
-      # Checkout GitHub Actions
-      - uses: actions/checkout@v5
-        with:
-          repository: kestra-io/actions
-          path: actions
-          ref: main
-
       # Setup build
-      - uses: ./actions/.github/actions/setup-build
+      - uses: kestra-io/actions/composite/setup-build@main
         id: build
         with:
           java-enabled: true
@@ -78,7 +71,6 @@ jobs:
           git checkout develop;
 
           if [[ "$RELEASE_VERSION" == *"-SNAPSHOT" ]]; then
-            # -SNAPSHOT qualifier maybe used to test release-candidates
             ./gradlew release -Prelease.useAutomaticVersion=true \
             -Prelease.releaseVersion="${RELEASE_VERSION}" \
             -Prelease.newVersion="${NEXT_VERSION}" \
@@ -89,4 +81,4 @@ jobs:
             -Prelease.releaseVersion="${RELEASE_VERSION}" \
             -Prelease.newVersion="${NEXT_VERSION}" \
             -Prelease.pushReleaseVersionBranch="${PUSH_RELEASE_BRANCH}"
-          fi
+          fi

.github/workflows/main.yml

@@ -3,6 +3,14 @@ name: Main Workflow
 on:
   workflow_dispatch:
     inputs:
+      skip-test:
+        description: 'Skip test'
+        type: choice
+        required: true
+        default: 'false'
+        options:
+          - "true"
+          - "false"
       plugin-version:
           description: "plugins version"
           required: false
@@ -24,13 +32,14 @@ jobs:
   tests:
     name: Execute tests
     uses: ./.github/workflows/workflow-test.yml
+    if: ${{ github.event.inputs.skip-test == 'false' || github.event.inputs.skip-test == '' }}
     with:
       report-status: false
 
   release:
     name: Release
     needs: [tests]
-    if: "!startsWith(github.ref, 'refs/heads/releases')"
+    if: "!failure() && !cancelled() && !startsWith(github.ref, 'refs/heads/releases')"
     uses: ./.github/workflows/workflow-release.yml
     with:
         plugin-version: ${{ inputs.plugin-version != '' && inputs.plugin-version || (github.ref == 'refs/heads/develop' && 'LATEST-SNAPSHOT' || 'LATEST') }}
@@ -44,13 +53,12 @@ jobs:
       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
     if: always()
-    env:
-      SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
     steps:
       - name: Trigger EE Workflow
         uses: peter-evans/repository-dispatch@v3
@@ -60,14 +68,9 @@ jobs:
           repository: kestra-io/kestra-ee
           event-type: "oss-updated"
 
-
       # Slack
       - name: Slack - Notification
-        uses: Gamesight/slack-workflow-status@master
-        if: ${{ always() && env.SLACK_WEBHOOK_URL != 0 }}
+        if: ${{ failure() && env.SLACK_WEBHOOK_URL != 0 && (github.ref == 'refs/heads/master' || github.ref == 'refs/heads/main' || github.ref == 'refs/heads/develop') }}
+        uses: kestra-io/actions/composite/slack-status@main
         with:
-          repo_token: ${{ secrets.GITHUB_TOKEN }}
-          slack_webhook_url: ${{ secrets.SLACK_WEBHOOK_URL }}
-          name: GitHub Actions
-          icon_emoji: ":github-actions:"
-          channel: "C02DQ1A7JLR" # _int_git channel
+          webhook-url: ${{ secrets.SLACK_WEBHOOK_URL }}

.github/workflows/pull-request.yml

@@ -60,19 +60,3 @@ jobs:
     name: E2E - Tests
     uses: ./.github/workflows/e2e.yml
 
-  end:
-    name: End
-    runs-on: ubuntu-latest
-    if: always()
-    needs: [frontend, backend]
-    steps:
-      # Slack
-      - name: Slack notification
-        uses: Gamesight/slack-workflow-status@master
-        if: ${{ always() && env.SLACK_WEBHOOK_URL != 0 }}
-        with:
-          repo_token: ${{ secrets.GITHUB_TOKEN }}
-          slack_webhook_url: ${{ secrets.SLACK_WEBHOOK_URL }}
-          name: GitHub Actions
-          icon_emoji: ":github-actions:"
-          channel: "C02DQ1A7JLR"

.github/workflows/setversion-tag.yml

@@ -34,11 +34,14 @@ jobs:
           fi
 
       # Checkout
-      - uses: actions/checkout@v5
+      - name: Checkout
+        uses: actions/checkout@v5
         with:
           fetch-depth: 0
+          token: ${{ secrets.GH_PERSONAL_TOKEN }}
 
-      - name: Configure Git
+      # Configure
+      - name: Git - Configure
         run: |
           git config --global user.email "41898282+github-actions[bot]@users.noreply.github.com"
           git config --global user.name "github-actions[bot]"
@@ -54,4 +57,4 @@ jobs:
           git commit -m"chore(version): update to version '$RELEASE_VERSION'"
           git push
           git tag -a "v$RELEASE_VERSION" -m"v$RELEASE_VERSION"
-          git push --tags
+          git push --tags

.github/workflows/vulnerabilities-check.yml

@@ -21,13 +21,6 @@ jobs:
         with:
           fetch-depth: 0
 
-      # Checkout GitHub Actions
-      - uses: actions/checkout@v5
-        with:
-          repository: kestra-io/actions
-          path: actions
-          ref: main
-
       # Setup build
       - uses: ./actions/.github/actions/setup-build
         id: build
@@ -70,15 +63,8 @@ jobs:
         with:
           fetch-depth: 0
 
-      # Checkout GitHub Actions
-      - uses: actions/checkout@v5
-        with:
-          repository: kestra-io/actions
-          path: actions
-          ref: main
-
       # Setup build
-      - uses: ./actions/.github/actions/setup-build
+      - uses: kestra-io/actions/composite/setup-build@main
         id: build
         with:
           java-enabled: false
@@ -87,7 +73,7 @@ jobs:
 
       # Run Trivy image scan for Docker vulnerabilities, see https://github.com/aquasecurity/trivy-action
       - name: Docker Vulnerabilities Check
-        uses: aquasecurity/trivy-action@0.32.0
+        uses: aquasecurity/trivy-action@0.33.1
         with:
           image-ref: kestra/kestra:develop
           format: 'template'
@@ -115,24 +101,16 @@ jobs:
         with:
           fetch-depth: 0
 
-      # Checkout GitHub Actions
-      - uses: actions/checkout@v5
-        with:
-          repository: kestra-io/actions
-          path: actions
-          ref: main
-
       # Setup build
-      - uses: ./actions/.github/actions/setup-build
+      - uses: kestra-io/actions/composite/setup-build@main
         id: build
         with:
           java-enabled: false
           node-enabled: false
-          caches-enabled: true
 
       # Run Trivy image scan for Docker vulnerabilities, see https://github.com/aquasecurity/trivy-action
       - name: Docker Vulnerabilities Check
-        uses: aquasecurity/trivy-action@0.32.0
+        uses: aquasecurity/trivy-action@0.33.1
         with:
           image-ref: kestra/kestra:latest
           format: table

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

@@ -20,6 +20,7 @@ permissions:
   contents: write
   checks: write
   actions: read
+  pull-requests: write
 
 jobs:
   test:
@@ -35,7 +36,7 @@ jobs:
           fetch-depth: 0
 
       # Setup build
-      - uses: kestra-io/actions/.github/actions/setup-build@main
+      - uses: kestra-io/actions/composite/setup-build@main
         name: Setup - Build
         id: build
         with:
@@ -59,84 +60,15 @@ jobs:
           export GOOGLE_APPLICATION_CREDENTIALS=$HOME/.gcp-service-account.json
           ./gradlew check javadoc --parallel
 
-      # report test
-      - name: Test - Publish Test Results
-        uses: dorny/test-reporter@v2
-        if: always()
-        with:
-          name: Java Tests Report
-          reporter: java-junit
-          path: '**/build/test-results/test/TEST-*.xml'
-          list-suites: 'failed'
-          list-tests: 'failed'
-          fail-on-error: 'false'
-          token: ${{ secrets.GITHUB_AUTH_TOKEN }}
-
-      # Sonar
-      - name: Test - Analyze with Sonar
-        if: env.SONAR_TOKEN != ''
+      - name: comment PR with test report
+        if: ${{ !cancelled() && github.event_name == 'pull_request' }}
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_AUTH_TOKEN }}
-          SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
-        shell: bash
-        run: ./gradlew sonar --info
+        run: npx --yes @kestra-io/kestra-devtools generateTestReportSummary --only-errors --ci $(pwd)
 
-      # GCP
-      - name: GCP - Auth with unit test account
-        id: auth
-        if: always() && env.GOOGLE_SERVICE_ACCOUNT != ''
-        continue-on-error: true
-        uses: "google-github-actions/auth@v2"
-        with:
-          credentials_json: "${{ secrets.GOOGLE_SERVICE_ACCOUNT }}"
-
-      - name: GCP - Setup Cloud SDK
-        if: env.GOOGLE_SERVICE_ACCOUNT != ''
-        uses: "google-github-actions/setup-gcloud@v2"
-
-        # Allure check
-      - 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
-
-      # Jacoco
-      - name: Jacoco - Copy reports
-        if: env.GOOGLE_SERVICE_ACCOUNT != ''
-        continue-on-error: true
-        shell: bash
-        run: |
-          mv build/reports/jacoco/testCodeCoverageReport build/reports/jacoco/test/
-          mv build/reports/jacoco/test/testCodeCoverageReport.xml build/reports/jacoco/test/jacocoTestReport.xml
-          gsutil -m rsync -d -r  build/reports/jacoco/test/ gs://internal-kestra-host/${{ format('{0}/{1}', github.repository, 'jacoco') }}
-
-      # Codecov
-      - name: Codecov - Upload coverage reports
-        uses: codecov/codecov-action@v5
+      # Report Java
+      - name: Report - Java
+        uses: kestra-io/actions/composite/report-java@main
         if: ${{ !cancelled() }}
-        continue-on-error: true
         with:
-          token: ${{ secrets.CODECOV_TOKEN }}
-          flags: backend
-
-      - name: Codecov - Upload test results
-        uses: codecov/test-results-action@v1
-        if: ${{ !cancelled() }}
-        continue-on-error: true
-        with:
-          token: ${{ secrets.CODECOV_TOKEN }}
-          flags: backend
+          secrets: ${{ toJSON(secrets) }}

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

@@ -26,7 +26,7 @@ jobs:
         run: npm ci
 
       # Setup build
-      - uses: kestra-io/actions/.github/actions/setup-build@main
+      - uses: kestra-io/actions/composite/setup-build@main
         name: Setup - Build
         id: build
         with:

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

@@ -25,15 +25,6 @@ jobs:
           fetch-depth: 0
           submodules: true
 
-      # Checkout GitHub Actions
-      - name: Checkout - Actions
-        uses: actions/checkout@v5
-        with:
-          repository: kestra-io/actions
-          sparse-checkout-cone-mode: true
-          path: actions
-          sparse-checkout: |
-            .github/actions
 
       # Download Exec
       # Must be done after checkout actions
@@ -59,7 +50,7 @@ jobs:
 
       # GitHub Release
       - name: Create GitHub release
-        uses: ./actions/.github/actions/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 }}
@@ -82,7 +73,7 @@ jobs:
 
       - name: Merge Release Notes
         if: ${{ startsWith(github.ref, 'refs/tags/v') }}
-        uses: ./actions/.github/actions/github-release-note-merge
+        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

@@ -11,6 +11,14 @@ on:
         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
@@ -179,6 +187,11 @@ jobs:
         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:
@@ -187,14 +200,9 @@ jobs:
     env:
       SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
     steps:
-
-      # Slack
       - name: Slack notification
-        uses: Gamesight/slack-workflow-status@master
-        if: ${{ always() && env.SLACK_WEBHOOK_URL != 0 }}
+        if: ${{ failure() && env.SLACK_WEBHOOK_URL != 0 }}
+        uses: kestra-io/actions/composite/slack-status@main
         with:
-          repo_token: ${{ secrets.GITHUB_TOKEN }}
-          slack_webhook_url: ${{ secrets.SLACK_WEBHOOK_URL }}
-          name: GitHub Actions
-          icon_emoji: ':github-actions:'
-          channel: 'C02DQ1A7JLR' # _int_git channel
+          webhook-url: ${{ secrets.SLACK_WEBHOOK_URL }}
+

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

@@ -29,7 +29,7 @@ jobs:
 
       # Setup build
       - name: Setup - Build
-        uses: kestra-io/actions/.github/actions/setup-build@main
+        uses: kestra-io/actions/composite/setup-build@main
         id: build
         with:
           java-enabled: true

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

@@ -7,7 +7,7 @@ on:
 jobs:
   publish:
     name: Pull Request - Delete Docker
-    if: github.repository == github.event.pull_request.head.repo.full_name # prevent running on forks
+    if: github.repository == 'kestra-io/kestra' # prevent running on forks
     runs-on: ubuntu-latest
     steps:
       - uses: dataaxiom/ghcr-cleanup-action@v1

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

@@ -8,12 +8,12 @@ on:
 jobs:
   build-artifacts:
     name: Build Artifacts
-    if: github.repository == github.event.pull_request.head.repo.full_name # prevent running on forks
+    if: github.repository == 'kestra-io/kestra' # prevent running on forks
     uses: ./.github/workflows/workflow-build-artifacts.yml
 
   publish:
     name: Publish Docker
-    if: github.repository == github.event.pull_request.head.repo.full_name # prevent running on forks
+    if: github.repository == 'kestra-io/kestra' # prevent running on forks
     runs-on: ubuntu-latest
     needs: build-artifacts
     env:
@@ -62,7 +62,7 @@ jobs:
 
       # Add comment on pull request
       - name: Add comment to PR
-        uses: actions/github-script@v7
+        uses: actions/github-script@v8
         with:
           github-token: ${{ secrets.GITHUB_TOKEN }}
           script: |

.github/workflows/workflow-test.yml

@@ -84,14 +84,12 @@ jobs:
     name: Notify - Slack
     runs-on: ubuntu-latest
     needs: [ frontend, backend ]
-    if: github.event_name == 'schedule'
     steps:
       - name: Notify failed CI
-        id: send-ci-failed
         if: |
-          always() && (needs.frontend.result != 'success' ||
-          needs.backend.result != 'success')
-        uses: kestra-io/actions/.github/actions/send-ci-failed@main
-        env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
+          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 }}

.plugins

@@ -36,6 +36,7 @@
 #plugin-gemini:io.kestra.plugin:plugin-gemini:LATEST
 #plugin-git:io.kestra.plugin:plugin-git:LATEST
 #plugin-github:io.kestra.plugin:plugin-github:LATEST
+#plugin-gitlab:io.kestra.plugin:plugin-gitlab:LATEST
 #plugin-googleworkspace:io.kestra.plugin:plugin-googleworkspace:LATEST
 #plugin-graalvm:io.kestra.plugin:plugin-graalvm:LATEST
 #plugin-graphql:io.kestra.plugin:plugin-graphql:LATEST
@@ -108,16 +109,17 @@
 #plugin-serdes:io.kestra.plugin:plugin-serdes:LATEST
 #plugin-servicenow:io.kestra.plugin:plugin-servicenow:LATEST
 #plugin-sifflet:io.kestra.plugin:plugin-sifflet:LATEST
-#plugin-singer:io.kestra.plugin:plugin-singer:LATEST
 #plugin-soda:io.kestra.plugin:plugin-soda:LATEST
 #plugin-solace:io.kestra.plugin:plugin-solace:LATEST
 #plugin-spark:io.kestra.plugin:plugin-spark:LATEST
 #plugin-sqlmesh:io.kestra.plugin:plugin-sqlmesh:LATEST
+#plugin-supabase:io.kestra.plugin:plugin-supabase:LATEST
 #plugin-surrealdb:io.kestra.plugin:plugin-surrealdb:LATEST
 #plugin-terraform:io.kestra.plugin:plugin-terraform:LATEST
 #plugin-transform:io.kestra.plugin:plugin-transform-grok:LATEST
 #plugin-transform:io.kestra.plugin:plugin-transform-json:LATEST
 #plugin-tika:io.kestra.plugin:plugin-tika:LATEST
+#plugin-trivy:io.kestra.plugin:plugin-trivy:LATEST
 #plugin-weaviate:io.kestra.plugin:plugin-weaviate:LATEST
 #plugin-zendesk:io.kestra.plugin:plugin-zendesk:LATEST
 #plugin-typesense:io.kestra.plugin:plugin-typesense:LATEST

Makefile

@@ -89,7 +89,7 @@ build-docker: build-exec
 		--compress \
 		--rm \
 		-f ./Dockerfile \
-		--build-arg="APT_PACKAGES=python3 python3-venv python-is-python3 python3-pip nodejs npm curl zip unzip jattach" \
+		--build-arg="APT_PACKAGES=python3 python-is-python3 python3-pip curl jattach" \
 		--build-arg="PYTHON_LIBRARIES=kestra" \
 		-t ${DOCKER_IMAGE}:${VERSION} ${DOCKER_PATH} || exit 1 ;
 

build.gradle

@@ -21,7 +21,7 @@ plugins {
 
     // test
     id "com.adarshr.test-logger" version "4.0.0"
-    id "org.sonarqube" version "6.2.0.5505"
+    id "org.sonarqube" version "6.3.1.5724"
     id 'jacoco-report-aggregation'
 
     // helper

cli/build.gradle

@@ -33,7 +33,11 @@ dependencies {
 
     implementation project(":storage-local")
 
+    // Kestra server components
+    implementation project(":executor")
+    implementation project(":scheduler")
     implementation project(":webserver")
+    implementation project(":worker")
 
     //test
     testImplementation "org.wiremock:wiremock-jetty12"

cli/src/main/java/io/kestra/cli/StandAloneRunner.java

@@ -1,15 +1,14 @@
-package io.kestra.core.runners;
+package io.kestra.cli;
 
-import io.kestra.core.schedulers.AbstractScheduler;
+import io.kestra.core.runners.*;
 import io.kestra.core.server.Service;
 import io.kestra.core.utils.Await;
 import io.kestra.core.utils.ExecutorsUtils;
+import io.kestra.worker.DefaultWorker;
 import io.micronaut.context.ApplicationContext;
-import io.micronaut.context.annotation.Requires;
 import io.micronaut.context.annotation.Value;
 import jakarta.annotation.PreDestroy;
 import jakarta.inject.Inject;
-import jakarta.inject.Singleton;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 
@@ -24,9 +23,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
 
 @SuppressWarnings("try")
 @Slf4j
-@Singleton
-@Requires(missingBeans = RunnerInterface.class)
-public class StandAloneRunner implements RunnerInterface, AutoCloseable {
+public class StandAloneRunner implements Runnable, AutoCloseable {
     @Setter protected int workerThread = Math.max(3, Runtime.getRuntime().availableProcessors());
     @Setter protected boolean schedulerEnabled = true;
     @Setter protected boolean workerEnabled = true;
@@ -45,7 +42,7 @@ public class StandAloneRunner implements RunnerInterface, AutoCloseable {
 
     private final AtomicBoolean running = new AtomicBoolean(false);
 
-    private volatile ExecutorService poolExecutor;
+    private ExecutorService poolExecutor;
 
     @Override
     public void run() {
@@ -57,20 +54,20 @@ public class StandAloneRunner implements RunnerInterface, AutoCloseable {
         if (workerEnabled) {
             // FIXME: For backward-compatibility with Kestra 0.15.x and earliest we still used UUID for Worker ID instead of IdUtils
             String workerID = UUID.randomUUID().toString();
-            Worker worker = applicationContext.createBean(Worker.class, workerID, workerThread, null);
+            Worker worker = applicationContext.createBean(DefaultWorker.class, workerID, workerThread, null);
             applicationContext.registerSingleton(worker); //
             poolExecutor.execute(worker);
             servers.add(worker);
         }
 
         if (schedulerEnabled) {
-            AbstractScheduler scheduler = applicationContext.getBean(AbstractScheduler.class);
+            Scheduler scheduler = applicationContext.getBean(Scheduler.class);
             poolExecutor.execute(scheduler);
             servers.add(scheduler);
         }
 
         if (indexerEnabled) {
-            IndexerInterface indexer = applicationContext.getBean(IndexerInterface.class);
+            Indexer indexer = applicationContext.getBean(Indexer.class);
             poolExecutor.execute(indexer);
             servers.add(indexer);
         }

cli/src/main/java/io/kestra/cli/commands/flows/FlowTestCommand.java

@@ -8,7 +8,7 @@ import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.repositories.LocalFlowRepositoryLoader;
 import io.kestra.core.runners.FlowInputOutput;
 import io.kestra.core.runners.RunnerUtils;
-import io.kestra.core.runners.StandAloneRunner;
+import io.kestra.cli.StandAloneRunner;
 import io.micronaut.context.ApplicationContext;
 import jakarta.inject.Inject;
 import jakarta.validation.ConstraintViolationException;
@@ -72,7 +72,6 @@ public class FlowTestCommand extends AbstractApiCommand {
     public Integer call() throws Exception {
         super.call();
 
-        StandAloneRunner runner = applicationContext.getBean(StandAloneRunner.class);
         LocalFlowRepositoryLoader repositoryLoader = applicationContext.getBean(LocalFlowRepositoryLoader.class);
         FlowRepositoryInterface flowRepository = applicationContext.getBean(FlowRepositoryInterface.class);
         FlowInputOutput flowInputOutput = applicationContext.getBean(FlowInputOutput.class);
@@ -89,7 +88,7 @@ public class FlowTestCommand extends AbstractApiCommand {
             inputs.put(this.inputs.get(i), this.inputs.get(i+1));
         }
 
-        try {
+        try (StandAloneRunner runner = applicationContext.createBean(StandAloneRunner.class);){
             runner.run();
             repositoryLoader.load(tenantService.getTenantId(tenantId), file.toFile());
 
@@ -103,8 +102,6 @@ public class FlowTestCommand extends AbstractApiCommand {
                 (flow, execution) -> flowInputOutput.readExecutionInputs(flow, execution, inputs),
                 Duration.ofHours(1)
             );
-
-            runner.close();
         } catch (ConstraintViolationException e) {
             throw new CommandLine.ParameterException(this.spec.commandLine(), e.getMessage());
         } catch (IOException | TimeoutException e) {

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.core.services.SkipExecutionService;
+import io.kestra.executor.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

@@ -2,7 +2,7 @@ package io.kestra.cli.commands.servers;
 
 import com.google.common.collect.ImmutableMap;
 import io.kestra.core.models.ServerType;
-import io.kestra.core.runners.IndexerInterface;
+import io.kestra.core.runners.Indexer;
 import io.kestra.core.utils.Await;
 import io.micronaut.context.ApplicationContext;
 import jakarta.inject.Inject;
@@ -29,7 +29,7 @@ public class IndexerCommand extends AbstractServerCommand {
     public Integer call() throws Exception {
         super.call();
 
-        IndexerInterface indexer = applicationContext.getBean(IndexerInterface.class);
+        Indexer indexer = applicationContext.getBean(Indexer.class);
         indexer.run();
 
         Await.until(() -> !this.applicationContext.isRunning());

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

@@ -2,7 +2,7 @@ package io.kestra.cli.commands.servers;
 
 import com.google.common.collect.ImmutableMap;
 import io.kestra.core.models.ServerType;
-import io.kestra.core.schedulers.AbstractScheduler;
+import io.kestra.scheduler.AbstractScheduler;
 import io.kestra.core.utils.Await;
 import io.micronaut.context.ApplicationContext;
 import jakarta.inject.Inject;

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

@@ -6,8 +6,8 @@ import io.kestra.cli.services.TenantIdSelectorService;
 import io.kestra.core.contexts.KestraContext;
 import io.kestra.core.models.ServerType;
 import io.kestra.core.repositories.LocalFlowRepositoryLoader;
-import io.kestra.core.runners.StandAloneRunner;
-import io.kestra.core.services.SkipExecutionService;
+import io.kestra.cli.StandAloneRunner;
+import io.kestra.executor.SkipExecutionService;
 import io.kestra.core.services.StartExecutorService;
 import io.kestra.core.utils.Await;
 import io.micronaut.context.ApplicationContext;
@@ -109,26 +109,27 @@ public class StandAloneCommand extends AbstractServerCommand {
             }
         }
 
-        StandAloneRunner standAloneRunner = applicationContext.getBean(StandAloneRunner.class);
+        try (StandAloneRunner standAloneRunner = applicationContext.getBean(StandAloneRunner.class)) {
 
-        if (this.workerThread == 0) {
-            standAloneRunner.setWorkerEnabled(false);
-        } else {
-            standAloneRunner.setWorkerThread(this.workerThread);
+            if (this.workerThread == 0) {
+                standAloneRunner.setWorkerEnabled(false);
+            } else {
+                standAloneRunner.setWorkerThread(this.workerThread);
+            }
+
+            if (this.indexerDisabled) {
+                standAloneRunner.setIndexerEnabled(false);
+            }
+
+            standAloneRunner.run();
+
+            if (fileWatcher != null) {
+                fileWatcher.startListeningFromConfig();
+            }
+
+            Await.until(() -> !this.applicationContext.isRunning());
         }
 
-        if (this.indexerDisabled) {
-            standAloneRunner.setIndexerEnabled(false);
-        }
-
-        standAloneRunner.run();
-
-        if (fileWatcher != null) {
-            fileWatcher.startListeningFromConfig();
-        }
-
-        Await.until(() -> !this.applicationContext.isRunning());
-
         return 0;
     }
 }

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

@@ -2,7 +2,7 @@ package io.kestra.cli.commands.servers;
 
 import com.google.common.collect.ImmutableMap;
 import io.kestra.core.models.ServerType;
-import io.kestra.core.runners.IndexerInterface;
+import io.kestra.core.runners.Indexer;
 import io.kestra.core.utils.Await;
 import io.kestra.core.utils.ExecutorsUtils;
 import io.micronaut.context.ApplicationContext;
@@ -54,7 +54,7 @@ public class WebServerCommand extends AbstractServerCommand {
         if (!indexerDisabled) {
             log.info("Starting an embedded indexer, this can be disabled by using `--no-indexer`.");
             poolExecutor = executorsUtils.cachedThreadPool("webserver-indexer");
-            poolExecutor.execute(applicationContext.getBean(IndexerInterface.class));
+            poolExecutor.execute(applicationContext.getBean(Indexer.class));
             shutdownHook(false, () -> poolExecutor.shutdown());
         }
 

cli/src/main/resources/application.yml

@@ -18,6 +18,10 @@ micronaut:
       root:
         paths: classpath:root
         mapping: /**
+  codec:
+    json:
+      additional-types:
+        - application/scim+json
   server:
     max-request-size: 10GB
     multipart:
@@ -78,8 +82,19 @@ micronaut:
       type: scheduled
       core-pool-size: 1
 
-  # Disable OpenTelemetry metrics by default, users that need it must enable it and configure the collector URL.
   metrics:
+    binders:
+      retry:
+        enabled: true
+      netty:
+        queues:
+          enabled: true
+        bytebuf-allocators:
+          enabled: true
+        channels:
+          enabled: true
+
+    # Disable OpenTelemetry metrics by default, users that need it must enable it and configure the collector URL.
     export:
       otlp:
         enabled: false
@@ -92,6 +107,8 @@ jackson:
   serialization-inclusion: non_null
   deserialization:
     FAIL_ON_UNKNOWN_PROPERTIES: false
+  mapper:
+    ACCEPT_CASE_INSENSITIVE_ENUMS: true
 
 endpoints:
   all:
@@ -100,6 +117,10 @@ endpoints:
     sensitive: false
   health:
     details-visible: ANONYMOUS
+    disk-space:
+      enabled: false
+    discovery-client:
+      enabled: false
   loggers:
     write-sensitive: false
   env:
@@ -133,12 +154,47 @@ kestra:
   tutorial-flows:
     # Automatically loads all tutorial flows at startup.
     enabled: true
+
   retries:
     attempts: 5
     multiplier: 2.0
     delay: 1s
     maxDelay: ""
 
+  server:
+    basic-auth:
+      # These URLs will not be authenticated, by default we open some of the Micronaut default endpoints but not all for security reasons
+      open-urls:
+        - "/ping"
+        - "/api/v1/executions/webhook/"
+        - "/api/v1/main/executions/webhook/"
+        - "/api/v1/*/executions/webhook/"
+        - "/api/v1/basicAuthValidationErrors"
+
+    preview:
+      initial-rows: 100
+      max-rows: 5000
+
+    # The expected time for this server to complete all its tasks before initiating a graceful shutdown.
+    terminationGracePeriod: 5m
+    workerTaskRestartStrategy: AFTER_TERMINATION_GRACE_PERIOD
+    # Configuration for Liveness and Heartbeat mechanism between servers.
+    liveness:
+      enabled: true
+      # The expected time between liveness probe.
+      interval: 10s
+      # The timeout used to detect service failures.
+      timeout: 1m
+      # The time to wait before executing a liveness probe.
+      initialDelay: 1m
+      # The expected time between service heartbeats.
+      heartbeatInterval: 3s
+    service:
+      purge:
+        initial-delay: 1h
+        fixed-delay: 1d
+        retention: 30d
+
   jdbc:
     queues:
       min-poll-interval: 25ms
@@ -150,7 +206,7 @@ kestra:
       fixed-delay: 1h
       retention: 7d
       types:
-        - type : io.kestra.core.models.executions.LogEntry
+        - type: io.kestra.core.models.executions.LogEntry
           retention: 1h
         - type: io.kestra.core.models.executions.MetricEntry
           retention: 1h
@@ -182,37 +238,12 @@ kestra:
   traces:
     root: DISABLED
 
-  server:
-    basic-auth:
-      # These URLs will not be authenticated, by default we open some of the Micronaut default endpoints but not all for security reasons
-      open-urls:
-        - "/ping"
-        - "/api/v1/executions/webhook/"
-    preview:
-      initial-rows: 100
-      max-rows: 5000
-    # The expected time for this server to complete all its tasks before initiating a graceful shutdown.
-    terminationGracePeriod: 5m
-    workerTaskRestartStrategy: AFTER_TERMINATION_GRACE_PERIOD
-    # Configuration for Liveness and Heartbeat mechanism between servers.
-    liveness:
-      enabled: true
-      # The expected time between liveness probe.
-      interval: 10s
-      # The timeout used to detect service failures.
-      timeout: 1m
-      # The time to wait before executing a liveness probe.
-      initialDelay: 1m
-      # The expected time between service heartbeats.
-      heartbeatInterval: 3s
-    service:
-      purge:
-        initial-delay: 1h
-        fixed-delay: 1d
-        retention: 30d
+  ui-anonymous-usage-report:
+    enabled: true
+
   anonymous-usage-report:
     enabled: true
-    uri: https://api.kestra.io/v1/server-events/
+    uri: https://api.kestra.io/v1/reports/server-events
     initial-delay: 5m
     fixed-delay: 1h
 

core/build.gradle

@@ -37,15 +37,15 @@ dependencies {
     implementation 'nl.basjes.gitignore:gitignore-reader'
     implementation group: 'dev.failsafe', name: 'failsafe'
     implementation 'com.github.ben-manes.caffeine:caffeine'
-    implementation 'com.github.ksuid:ksuid:1.1.3'
+    implementation 'com.github.ksuid:ksuid:1.1.4'
     api 'org.apache.httpcomponents.client5:httpclient5'
 
     // plugins
     implementation 'org.apache.maven.resolver:maven-resolver-impl'
-    implementation 'org.apache.maven.resolver:maven-resolver-supplier'
+    implementation 'org.apache.maven.resolver:maven-resolver-supplier-mvn3'
     implementation 'org.apache.maven.resolver:maven-resolver-connector-basic'
     implementation 'org.apache.maven.resolver:maven-resolver-transport-file'
-    implementation 'org.apache.maven.resolver:maven-resolver-transport-http'
+    implementation 'org.apache.maven.resolver:maven-resolver-transport-apache'
 
     // scheduler
     implementation group: 'com.cronutils', name: 'cron-utils'
@@ -63,6 +63,10 @@ dependencies {
         exclude group: 'com.fasterxml.jackson.core'
     }
 
+    // micrometer
+    implementation "io.micronaut.micrometer:micronaut-micrometer-observation"
+    implementation 'io.micrometer:micrometer-java21'
+
     // test
     testAnnotationProcessor project(':processor')
     testImplementation project(':tests')
@@ -70,6 +74,9 @@ dependencies {
     testImplementation project(':repository-memory')
     testImplementation project(':runner-memory')
     testImplementation project(':storage-local')
+    testImplementation project(':worker')
+    testImplementation project(':scheduler')
+    testImplementation project(':executor')
 
     testImplementation "io.micronaut:micronaut-http-client"
     testImplementation "io.micronaut:micronaut-http-server-netty"

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

@@ -36,6 +36,7 @@ public class Plugin {
     private List<PluginElementMetadata> appBlocks;
     private List<PluginElementMetadata> charts;
     private List<PluginElementMetadata> dataFilters;
+    private List<PluginElementMetadata> dataFiltersKPI;
     private List<PluginElementMetadata> logExporters;
     private List<PluginElementMetadata> additionalPlugins;
     private List<PluginSubGroup.PluginCategory> categories;
@@ -96,6 +97,7 @@ public class Plugin {
         plugin.appBlocks = filterAndGetTypeWithMetadata(registeredPlugin.getAppBlocks(), packagePredicate);
         plugin.charts = filterAndGetTypeWithMetadata(registeredPlugin.getCharts(), packagePredicate);
         plugin.dataFilters = filterAndGetTypeWithMetadata(registeredPlugin.getDataFilters(), packagePredicate);
+        plugin.dataFiltersKPI = filterAndGetTypeWithMetadata(registeredPlugin.getDataFiltersKPI(), packagePredicate);
         plugin.logExporters = filterAndGetTypeWithMetadata(registeredPlugin.getLogExporters(), packagePredicate);
         plugin.additionalPlugins = filterAndGetTypeWithMetadata(registeredPlugin.getAdditionalPlugins(), packagePredicate);
 

core/src/main/java/io/kestra/core/events/CrudEvent.java

@@ -3,30 +3,88 @@ package io.kestra.core.events;
 import io.micronaut.core.annotation.Nullable;
 import io.micronaut.http.HttpRequest;
 import io.micronaut.http.context.ServerRequestContext;
-import lombok.AllArgsConstructor;
 import lombok.Getter;
 
-@AllArgsConstructor
+import java.util.Objects;
+
 @Getter
 public class CrudEvent<T> {
-    T model;
+    private final T model;
     @Nullable
-    T previousModel;
-    CrudEventType type;
-    HttpRequest<?> request;
-
+    private final T previousModel;
+    private final CrudEventType type;
+    private final HttpRequest<?> request;
+    
+    /**
+     * Static helper method for creating a new {@link CrudEventType#UPDATE} CrudEvent.
+     *
+     * @param model the new created model.
+     * @param <T>   type of the model.
+     * @return the new {@link CrudEvent}.
+     */
+    public static <T> CrudEvent<T> create(T model) {
+        Objects.requireNonNull(model, "Can't create CREATE event with a null model");
+        return new CrudEvent<>(model, null, CrudEventType.CREATE);
+    }
+    
+    /**
+     * Static helper method for creating a new {@link CrudEventType#DELETE} CrudEvent.
+     *
+     * @param model the deleted model.
+     * @param <T>   type of the model.
+     * @return the new {@link CrudEvent}.
+     */
+    public static <T> CrudEvent<T> delete(T model) {
+        Objects.requireNonNull(model, "Can't create DELETE event with a null model");
+        return new CrudEvent<>(null, model, CrudEventType.DELETE);
+    }
+    
+    /**
+     * Static helper method for creating a new CrudEvent.
+     *
+     * @param before the model before the update.
+     * @param after  the model after the update.
+     * @param <T>   type of the model.
+     * @return the new {@link CrudEvent}.
+     */
+    public static <T> CrudEvent<T> of(T before, T after) {
+        
+        if (before == null && after == null) {
+            throw new IllegalArgumentException("Both before and after cannot be null");
+        }
+        
+        if (before == null) {
+            return create(after);
+        }
+        
+        if (after == null) {
+            return delete(before);
+        }
+        
+        return new CrudEvent<>(after, before, CrudEventType.UPDATE);
+    }
+    
+    /**
+     * @deprecated use the static factory methods.
+     */
+    @Deprecated
     public CrudEvent(T model, CrudEventType type) {
-        this.model = model;
-        this.type = type;
-        this.previousModel = null;
-        this.request = ServerRequestContext.currentRequest().orElse(null);
+        this(
+            CrudEventType.DELETE.equals(type) ? null : model,
+            CrudEventType.DELETE.equals(type) ? model : null, 
+            type, 
+            ServerRequestContext.currentRequest().orElse(null)
+        );
     }
 
     public CrudEvent(T model, T previousModel, CrudEventType type) {
+        this(model, previousModel, type, ServerRequestContext.currentRequest().orElse(null));
+    }
+    
+    public CrudEvent(T model, T previousModel, CrudEventType type, HttpRequest<?> request) {
         this.model = model;
         this.previousModel = previousModel;
         this.type = type;
-        this.request = ServerRequestContext.currentRequest().orElse(null);
+        this.request = request;
     }
-
 }

core/src/main/java/io/kestra/core/http/client/HttpClient.java

@@ -6,16 +6,24 @@ import io.kestra.core.http.HttpRequest;
 import io.kestra.core.http.HttpResponse;
 import io.kestra.core.http.client.apache.*;
 import io.kestra.core.http.client.configurations.HttpConfiguration;
+import io.kestra.core.runners.DefaultRunContext;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.serializers.JacksonMapper;
+import io.micrometer.common.KeyValues;
+import io.micrometer.core.instrument.binder.httpcomponents.hc5.ApacheHttpClientContext;
+import io.micrometer.core.instrument.binder.httpcomponents.hc5.DefaultApacheHttpClientObservationConvention;
+import io.micrometer.core.instrument.binder.httpcomponents.hc5.ObservationExecChainHandler;
+import io.micrometer.observation.ObservationRegistry;
 import io.micronaut.http.MediaType;
 import jakarta.annotation.Nullable;
 import lombok.Builder;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.ArrayUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hc.client5.http.ContextBuilder;
 import org.apache.hc.client5.http.auth.*;
 import org.apache.hc.client5.http.config.ConnectionConfig;
+import org.apache.hc.client5.http.impl.ChainElement;
 import org.apache.hc.client5.http.impl.DefaultAuthenticationStrategy;
 import org.apache.hc.client5.http.impl.auth.BasicCredentialsProvider;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
@@ -30,7 +38,6 @@ import org.apache.hc.core5.http.io.HttpClientResponseHandler;
 import org.apache.hc.core5.http.io.entity.EntityUtils;
 import org.apache.hc.core5.ssl.SSLContexts;
 import org.apache.hc.core5.util.Timeout;
-import org.codehaus.plexus.util.StringUtils;
 
 import java.io.Closeable;
 import java.io.IOException;
@@ -50,11 +57,16 @@ public class HttpClient implements Closeable {
     private transient CloseableHttpClient client;
     private final RunContext runContext;
     private final HttpConfiguration configuration;
+    private ObservationRegistry observationRegistry;
 
     @Builder
     public HttpClient(RunContext runContext, @Nullable HttpConfiguration configuration) throws IllegalVariableEvaluationException {
         this.runContext = runContext;
         this.configuration = configuration == null ? HttpConfiguration.builder().build() : configuration;
+        if (runContext instanceof DefaultRunContext defaultRunContext) {
+            this.observationRegistry = defaultRunContext.getApplicationContext().findBean(ObservationRegistry.class).orElse(null);
+        }
+
         this.client = this.createClient();
     }
 
@@ -67,6 +79,13 @@ public class HttpClient implements Closeable {
             .disableDefaultUserAgent()
             .setUserAgent("Kestra");
 
+        if (observationRegistry != null) {
+            // micrometer, must be placed before the retry strategy (see https://docs.micrometer.io/micrometer/reference/reference/httpcomponents.html#_retry_strategy_considerations)
+            builder.addExecInterceptorAfter(ChainElement.RETRY.name(), "micrometer",
+                new ObservationExecChainHandler(observationRegistry, new CustomApacheHttpClientObservationConvention())
+            );
+        }
+
         // logger
         if (this.configuration.getLogs() != null && this.configuration.getLogs().length > 0) {
             if (ArrayUtils.contains(this.configuration.getLogs(), HttpConfiguration.LoggingType.REQUEST_HEADERS) ||
@@ -297,4 +316,14 @@ public class HttpClient implements Closeable {
             this.client.close();
         }
     }
+
+    public static class CustomApacheHttpClientObservationConvention extends DefaultApacheHttpClientObservationConvention {
+        @Override
+        public KeyValues getLowCardinalityKeyValues(ApacheHttpClientContext context) {
+            return KeyValues.concat(
+                super.getLowCardinalityKeyValues(context),
+                KeyValues.of("type", "core-client")
+            );
+        }
+    }
 }

core/src/main/java/io/kestra/core/metrics/MeterRegistryBinderFactory.java

@@ -0,0 +1,34 @@
+package io.kestra.core.metrics;
+
+import io.micrometer.core.instrument.binder.jvm.JvmThreadDeadlockMetrics;
+import io.micrometer.java21.instrument.binder.jdk.VirtualThreadMetrics;
+import io.micronaut.configuration.metrics.annotation.RequiresMetrics;
+import io.micronaut.context.annotation.Bean;
+import io.micronaut.context.annotation.Factory;
+import io.micronaut.context.annotation.Primary;
+import io.micronaut.context.annotation.Requires;
+import jakarta.inject.Singleton;
+
+import static io.micronaut.configuration.metrics.micrometer.MeterRegistryFactory.MICRONAUT_METRICS_BINDERS;
+import static io.micronaut.core.util.StringUtils.FALSE;
+
+@Factory
+@RequiresMetrics
+
+public class MeterRegistryBinderFactory {
+    @Bean
+    @Primary
+    @Singleton
+    @Requires(property = MICRONAUT_METRICS_BINDERS + ".jvm.enabled", notEquals = FALSE)
+    public VirtualThreadMetrics virtualThreadMetrics() {
+        return new VirtualThreadMetrics();
+    }
+
+    @Bean
+    @Primary
+    @Singleton
+    @Requires(property = MICRONAUT_METRICS_BINDERS + ".jvm.enabled", notEquals = FALSE)
+    public JvmThreadDeadlockMetrics threadDeadlockMetricsMetrics() {
+        return new JvmThreadDeadlockMetrics();
+    }
+}

core/src/main/java/io/kestra/core/metrics/MetricRegistry.java

@@ -6,7 +6,6 @@ import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.models.triggers.TriggerContext;
 import io.kestra.core.runners.*;
-import io.kestra.core.schedulers.SchedulerExecutionWithTrigger;
 import io.micrometer.core.instrument.*;
 import io.micrometer.core.instrument.binder.MeterBinder;
 import io.micrometer.core.instrument.search.Search;
@@ -395,19 +394,6 @@ public class MetricRegistry {
         return triggerContext.getTenantId() == null ? baseTags : ArrayUtils.addAll(baseTags, TAG_TENANT_ID, triggerContext.getTenantId());
     }
 
-    /**
-     * Return tags for current {@link SchedulerExecutionWithTrigger}.
-     *
-     * @param schedulerExecutionWithTrigger the current SchedulerExecutionWithTrigger
-     * @return tags to apply to metrics
-     */
-    public String[] tags(SchedulerExecutionWithTrigger schedulerExecutionWithTrigger, String... tags) {
-        return ArrayUtils.addAll(
-            this.tags(schedulerExecutionWithTrigger.getExecution()),
-            tags
-        );
-    }
-
     /**
      * Return tags for current {@link ExecutionKilled}
      *

core/src/main/java/io/kestra/core/models/flows/input/MultiselectInput.java

@@ -3,6 +3,7 @@ package io.kestra.core.models.flows.input;
 import io.kestra.core.models.flows.Input;
 import io.kestra.core.models.flows.RenderableInput;
 import io.kestra.core.models.flows.Type;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.validations.ManualConstraintViolation;
 import io.kestra.core.validations.Regex;
 import io.swagger.v3.oas.annotations.media.Schema;
@@ -13,6 +14,7 @@ import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.experimental.SuperBuilder;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
@@ -56,6 +58,23 @@ public class MultiselectInput extends Input<List<String>> implements ItemTypeInt
     @Builder.Default
     Boolean allowCustomValue = false;
 
+    @Schema(
+        title = "Whether the first value of the multi-select should be selected by default."
+    )
+    @NotNull
+    @Builder.Default
+    Boolean autoSelectFirst = false;
+
+    @Override
+    public Property<List<String>> getDefaults() {
+        Property<List<String>> baseDefaults = super.getDefaults();
+        if (baseDefaults == null && autoSelectFirst && !Optional.ofNullable(values).map(Collection::isEmpty).orElse(true)) {
+            return Property.ofValue(List.of(values.getFirst()));
+        }
+
+        return baseDefaults;
+    }
+
     @Override
     public void validate(List<String> inputs) throws ConstraintViolationException {
         if (values != null && options != null) {
@@ -100,6 +119,7 @@ public class MultiselectInput extends Input<List<String>> implements ItemTypeInt
                 .dependsOn(getDependsOn())
                 .itemType(getItemType())
                 .displayName(getDisplayName())
+                .autoSelectFirst(getAutoSelectFirst())
                 .build();
         }
         return this;

core/src/main/java/io/kestra/core/models/flows/input/SelectInput.java

@@ -2,6 +2,7 @@ package io.kestra.core.models.flows.input;
 
 import io.kestra.core.models.flows.Input;
 import io.kestra.core.models.flows.RenderableInput;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.validations.ManualConstraintViolation;
 import io.kestra.core.validations.Regex;
 import io.swagger.v3.oas.annotations.media.Schema;
@@ -12,6 +13,7 @@ import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.experimental.SuperBuilder;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
@@ -46,6 +48,23 @@ public class SelectInput extends Input<String> implements RenderableInput {
     @Builder.Default
     Boolean isRadio = false;
 
+    @Schema(
+        title = "Whether the first value of the select should be selected by default."
+    )
+    @NotNull
+    @Builder.Default
+    Boolean autoSelectFirst = false;
+
+    @Override
+    public Property<String> getDefaults() {
+        Property<String> baseDefaults = super.getDefaults();
+        if (baseDefaults == null && autoSelectFirst && !Optional.ofNullable(values).map(Collection::isEmpty).orElse(true)) {
+            return Property.ofValue(values.getFirst());
+        }
+
+        return baseDefaults;
+    }
+
     @Override
     public void validate(String input) throws ConstraintViolationException {
         if (!values.contains(input) && this.getRequired()) {
@@ -78,6 +97,7 @@ public class SelectInput extends Input<String> implements RenderableInput {
                 .dependsOn(getDependsOn())
                 .displayName(getDisplayName())
                 .isRadio(getIsRadio())
+                .autoSelectFirst(getAutoSelectFirst())
                 .build();
         }
         return this;

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/models/triggers/multipleflows/MultipleCondition.java

@@ -3,7 +3,6 @@ package io.kestra.core.models.triggers.multipleflows;
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.models.conditions.Condition;
 import io.kestra.core.models.conditions.ConditionContext;
-import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.triggers.TimeWindow;
 import io.kestra.core.utils.Rethrow;
 import org.slf4j.Logger;
@@ -24,7 +23,7 @@ public interface MultipleCondition extends Rethrow.PredicateChecked<ConditionCon
 
     /**
      * This conditions will only validate previously calculated value on
-     * {@link io.kestra.core.services.FlowTriggerService#computeExecutionsFromFlowTriggers(Execution, List, Optional)}} and {@link MultipleConditionStorageInterface#save(List)} by the executor.
+     * io.kestra.executor.FlowTriggerService#computeExecutionsFromFlowTriggers(Execution, List, Optional) and {@link MultipleConditionStorageInterface#save(List)} by the executor.
      * The real validation is done here.
      */
     @Override

core/src/main/java/io/kestra/core/plugins/DefaultPluginRegistry.java

@@ -43,7 +43,7 @@ public class DefaultPluginRegistry implements PluginRegistry {
         static final DefaultPluginRegistry INSTANCE = new DefaultPluginRegistry();
     }
 
-    private final Map<PluginIdentifier, PluginClassAndMetadata<? extends Plugin>> pluginClassByIdentifier = new ConcurrentHashMap<>();
+    protected final Map<PluginIdentifier, PluginClassAndMetadata<? extends Plugin>> pluginClassByIdentifier = new ConcurrentHashMap<>();
     private final Map<PluginBundleIdentifier, RegisteredPlugin> plugins = new ConcurrentHashMap<>();
     private final PluginScanner scanner = new PluginScanner(DefaultPluginRegistry.class.getClassLoader());
     private final AtomicBoolean initialized = new AtomicBoolean(false);
@@ -103,11 +103,13 @@ public class DefaultPluginRegistry implements PluginRegistry {
      */
     @Override
     public void registerIfAbsent(final Path pluginPath) {
+        long start = System.currentTimeMillis();
         if (isPluginPathValid(pluginPath) && !isPluginPathScanned(pluginPath)) {
             List<RegisteredPlugin> scanned = scanner.scan(pluginPath);
             scanned.forEach(this::register);
             scannedPluginPaths.add(pluginPath);
         }
+        log.debug("Registered if absent plugins from path {} in {} ms", pluginPath, System.currentTimeMillis() - start);
     }
 
     private boolean isPluginPathScanned(final Path pluginPath) {
@@ -119,10 +121,12 @@ public class DefaultPluginRegistry implements PluginRegistry {
      */
     @Override
     public void register(final Path pluginPath) {
+        long start = System.currentTimeMillis();
         if (isPluginPathValid(pluginPath)) {
             List<RegisteredPlugin> scanned = scanner.scan(pluginPath);
             scanned.forEach(this::register);
         }
+        log.debug("Registered plugins from path {} in {} ms", pluginPath, System.currentTimeMillis() - start);
     }
 
     /**
@@ -191,20 +195,27 @@ public class DefaultPluginRegistry implements PluginRegistry {
      */
     public void register(final RegisteredPlugin plugin) {
         final PluginBundleIdentifier identifier = PluginBundleIdentifier.of(plugin);
-
-        // Skip registration if plugin-bundle already exists in the registry.
-        if (containsPluginBundle(identifier)) {
-            return;
+        // Skip registration if the same plugin already exists in the registry.
+        final RegisteredPlugin existing = plugins.get(identifier);
+        if (existing != null && existing.crc32() == plugin.crc32()) {
+            return; // same plugin already registered
         }
-
+        
         lock.lock();
         try {
+            if (existing != null) {
+                unregister(List.of(existing));
+            }
             plugins.put(PluginBundleIdentifier.of(plugin), plugin);
-            pluginClassByIdentifier.putAll(getPluginClassesByIdentifier(plugin));
+            registerAll(getPluginClassesByIdentifier(plugin));
         } finally {
             lock.unlock();
         }
     }
+    
+    protected void registerAll(Map<PluginIdentifier, PluginClassAndMetadata<? extends Plugin>> plugins) {
+        pluginClassByIdentifier.putAll(plugins);
+    }
 
     @SuppressWarnings("unchecked")
     protected Map<PluginIdentifier, PluginClassAndMetadata<? extends Plugin>> getPluginClassesByIdentifier(final RegisteredPlugin plugin) {

core/src/main/java/io/kestra/core/plugins/ExternalPlugin.java

@@ -6,6 +6,12 @@ import lombok.Getter;
 import lombok.ToString;
 
 import java.net.URL;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Enumeration;
+import java.util.jar.JarEntry;
+import java.util.jar.JarFile;
+import java.util.zip.CRC32;
 
 @AllArgsConstructor
 @Getter
@@ -14,4 +20,59 @@ import java.net.URL;
 public class ExternalPlugin {
     private final URL location;
     private final URL[] resources;
+    private volatile Long crc32; // lazy-val
+    
+    public ExternalPlugin(URL location, URL[] resources) {
+        this.location = location;
+        this.resources = resources;
+    }
+    
+    public Long getCrc32() {
+        if (this.crc32 == null) {
+            synchronized (this) {
+                if (this.crc32 == null) {
+                    this.crc32 = computeJarCrc32(location);
+                }
+            }
+        }
+        return crc32;
+    }
+    
+    /**
+     * Compute a CRC32 of the JAR File without reading the whole file
+     *
+     * @param location of the JAR File.
+     * @return the CRC32 of {@code -1} if the checksum can't be computed.
+     */
+    private static long computeJarCrc32(final URL location) {
+        CRC32 crc = new CRC32();
+        try (JarFile jar = new JarFile(location.toURI().getPath(), false)) {
+            Enumeration<JarEntry> entries = jar.entries();
+            byte[] buffer = new byte[Long.BYTES]; // reusable buffer to avoid re-allocation
+            
+            while (entries.hasMoreElements()) {
+                JarEntry entry = entries.nextElement();
+                crc.update(entry.getName().getBytes(StandardCharsets.UTF_8));
+                updateCrc32WithLong(crc, buffer, entry.getSize());
+                updateCrc32WithLong(crc, buffer, entry.getCrc());
+            }
+            
+            return crc.getValue();
+        } catch (Exception e) {
+            return -1;
+        }
+    }
+    
+    private static void updateCrc32WithLong(CRC32 crc32, byte[] reusable, long val) {
+        // fast long -> byte conversion
+        reusable[0] = (byte) (val >>> 56);
+        reusable[1] = (byte) (val >>> 48);
+        reusable[2] = (byte) (val >>> 40);
+        reusable[3] = (byte) (val >>> 32);
+        reusable[4] = (byte) (val >>> 24);
+        reusable[5] = (byte) (val >>> 16);
+        reusable[6] = (byte) (val >>> 8);
+        reusable[7] = (byte) val;
+        crc32.update(reusable);;
+    }
 }

core/src/main/java/io/kestra/core/plugins/PluginClassLoader.java

@@ -46,6 +46,7 @@ public class PluginClassLoader extends URLClassLoader {
         + "|dev.failsafe"
         + "|reactor"
         + "|io.opentelemetry"
+        + "|io.netty"
         + ")\\..*$");
 
     private final ClassLoader parent;

core/src/main/java/io/kestra/core/plugins/PluginRegistry.java

@@ -2,10 +2,14 @@ package io.kestra.core.plugins;
 
 import io.kestra.core.models.Plugin;
 
+import java.net.URL;
+import java.nio.ByteBuffer;
 import java.nio.file.Path;
 import java.util.List;
 import java.util.Optional;
 import java.util.function.Predicate;
+import java.util.zip.CRC32;
+import java.util.zip.Checksum;
 
 /**
  * Registry for managing all Kestra's {@link Plugin}.
@@ -123,4 +127,24 @@ public interface PluginRegistry {
      * @return {@code true} if supported. Otherwise {@code false}.
      */
     boolean isVersioningSupported();
+    
+    /**
+     * Computes a CRC32 hash value representing the current content of the plugin registry.
+     *
+     * @return a {@code long} containing the CRC32 checksum value, serving as a compact
+     *         representation of the registry's content
+     */
+    default long hash() {
+        Checksum crc32 = new CRC32();
+        
+        for (RegisteredPlugin plugin : plugins()) {
+            Optional.ofNullable(plugin.getExternalPlugin())
+                .map(ExternalPlugin::getCrc32)
+                .ifPresent(checksum -> {
+                    byte[] bytes = ByteBuffer.allocate(Long.BYTES).putLong(checksum).array();
+                    crc32.update(bytes, 0, bytes.length);
+                });
+        }
+        return crc32.getValue();
+    }
 }

core/src/main/java/io/kestra/core/plugins/PluginResolver.java

@@ -5,11 +5,15 @@ import lombok.extern.slf4j.Slf4j;
 import java.io.IOException;
 import java.net.MalformedURLException;
 import java.net.URL;
+import java.nio.ByteBuffer;
 import java.nio.file.DirectoryStream;
 import java.nio.file.Files;
 import java.nio.file.InvalidPathException;
 import java.nio.file.Path;
 import java.util.*;
+import java.util.jar.JarEntry;
+import java.util.jar.JarFile;
+import java.util.zip.CRC32;
 
 @Slf4j
 public class PluginResolver {
@@ -119,4 +123,5 @@ public class PluginResolver {
 
         return urls;
     }
+
 }

core/src/main/java/io/kestra/core/plugins/RegisteredPlugin.java

@@ -308,6 +308,10 @@ public class RegisteredPlugin {
         }
         return null;
     }
+    
+    public long crc32() {
+        return Optional.ofNullable(externalPlugin).map(ExternalPlugin::getCrc32).orElse(-1L);
+    }
 
     @Override
     public String toString() {

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

@@ -11,6 +11,10 @@ import lombok.Getter;
 import java.util.ArrayList;
 import java.util.List;
 
+// TODO for 2.0: this class is used as a queue consumer (which should have been the ExecutorInterface instead),
+//  a queue message (only in Kafka) and an execution context.
+//  At some point, we should rename it to ExecutorContext and move it to the executor module,
+//  then rename the ExecutorInterface to just Executor (to be used as a queue consumer)
 @Getter
 @AllArgsConstructor
 public class Executor {

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

@@ -1,7 +1,7 @@
 package io.kestra.core.runners;
 
-import java.io.Closeable;
+import io.kestra.core.server.Service;
 
-public interface ExecutorInterface extends Closeable, Runnable {
+public interface ExecutorInterface extends Service, Runnable {
 
 }

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

@@ -382,6 +382,15 @@ public class FlowInputOutput {
             .stream()
             .collect(HashMap::new, (m, v) -> m.put(v.getKey(), v.getValue().value()), HashMap::putAll)
         );
+        // Hack: Pre-inject all inputs that have a default value with 'null' to prevent
+        // RunContextFactory from attempting to render them when absent, which could
+        // otherwise cause an exception if a Pebble expression is involved.
+        List<Input<?>> inputs = Optional.ofNullable(flow).map(FlowInterface::getInputs).orElse(List.of());
+        for (Input<?> input : inputs) {
+            if (input.getDefaults() != null && !flattenInputs.containsKey(input.getId())) {
+                flattenInputs.put(input.getId(), null);
+            }
+        }
         return runContextFactory.of(flow, execution, vars -> vars.withInputs(flattenInputs));
     }
 

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

@@ -1,4 +1,7 @@
 package io.kestra.core.runners;
-// NOTE: this class is not used anymore but must be kept as it is used in as queue consumer both in JDBC and Kafka
-public class Indexer {
+
+import io.kestra.core.server.Service;
+
+public interface Indexer extends Service, Runnable {
+
 }

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

@@ -20,9 +20,11 @@ import io.kestra.core.queues.QueueInterface;
 import jakarta.annotation.Nullable;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.io.IOUtils;
 import org.slf4j.LoggerFactory;
 
 import java.io.*;
+import java.nio.charset.StandardCharsets;
 import java.time.Instant;
 import java.util.*;
 import java.util.function.Supplier;
@@ -142,8 +144,9 @@ public class RunContextLogger implements Supplier<org.slf4j.Logger> {
     }
 
     public void usedSecret(String secret) {
-        if (secret != null) {
+        if (secret != null && !secret.isEmpty()) {
             this.useSecrets.add(secret);
+            this.useSecrets.add(Base64.getEncoder().encodeToString(secret.getBytes(StandardCharsets.UTF_8)));
         }
     }
 

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

@@ -10,6 +10,7 @@ import io.kestra.core.models.flows.FlowInterface;
 import io.kestra.core.models.flows.Input;
 import io.kestra.core.models.flows.State;
 import io.kestra.core.models.flows.input.SecretInput;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.property.PropertyContext;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.triggers.AbstractTrigger;
@@ -282,15 +283,15 @@ public final class RunVariables {
 
                 if (flow != null && flow.getInputs() != null) {
                     // we add default inputs value from the flow if not already set, this will be useful for triggers
-                        flow.getInputs().stream()
-                            .filter(input -> input.getDefaults() != null && !inputs.containsKey(input.getId()))
-                            .forEach(input -> {
-                                try {
-                                    inputs.put(input.getId(), FlowInputOutput.resolveDefaultValue(input, propertyContext));
-                                } catch (IllegalVariableEvaluationException e) {
-                                    throw new RuntimeException("Unable to inject default value for input '" + input.getId() + "'", e);
-                                }
-                            });
+                    flow.getInputs().stream()
+                        .filter(input -> input.getDefaults() != null && !inputs.containsKey(input.getId()))
+                        .forEach(input -> {
+                            try {
+                                inputs.put(input.getId(), FlowInputOutput.resolveDefaultValue(input, propertyContext));
+                            } catch (IllegalVariableEvaluationException e) {
+                                // Silent catch, if an input depends on another input, or a variable that is populated at runtime / input filling time, we can't resolve it here.
+                            }
+                        });
                 }
 
                 if (!inputs.isEmpty()) {

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

@@ -1,5 +0,0 @@
-package io.kestra.core.runners;
-
-public interface RunnerInterface {
-    void run();
-}

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

@@ -1,4 +1,4 @@
-package io.kestra.core.schedulers;
+package io.kestra.core.runners;
 
 import java.util.function.Consumer;
 

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

@@ -2,6 +2,5 @@ package io.kestra.core.runners;
 
 import io.kestra.core.server.Service;
 
-public interface IndexerInterface extends Service, Runnable {
-
+public interface Scheduler extends Service, Runnable {
 }

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

@@ -1,4 +1,4 @@
-package io.kestra.core.schedulers;
+package io.kestra.core.runners;
 
 import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.flows.Flow;

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

@@ -45,7 +45,7 @@ final class Secret {
         for (var entry: data.entrySet()) {
             if (entry.getValue() instanceof Map map) {
                 // if some value are of type EncryptedString we decode them and replace the object
-                if (EncryptedString.TYPE.equalsIgnoreCase((String)map.get("type"))) {
+                if (map.get("type") instanceof String typeStr && EncryptedString.TYPE.equalsIgnoreCase(typeStr)) {
                     try {
                         String decoded = decrypt((String) map.get("value"));
                         decryptedMap.put(entry.getKey(), decoded);

core/src/main/java/io/kestra/core/schedulers/Scheduler.java

@@ -1,9 +0,0 @@
-package io.kestra.core.schedulers;
-
-import jakarta.inject.Singleton;
-
-@SuppressWarnings("try")
-@Singleton
-public interface Scheduler extends Runnable, AutoCloseable {
-
-}

core/src/main/java/io/kestra/core/serializers/JacksonMapper.java

@@ -163,31 +163,28 @@ public final class JacksonMapper {
             .build();
     }
 
-    public static Pair<JsonNode, JsonNode> getBiDirectionalDiffs(Object previous, Object current)  {
-        JsonNode previousJson = MAPPER.valueToTree(previous);
-        JsonNode newJson = MAPPER.valueToTree(current);
+    public static Pair<JsonNode, JsonNode> getBiDirectionalDiffs(Object before, Object after)  {
+        JsonNode beforeNode = MAPPER.valueToTree(before);
+        JsonNode afterNode = MAPPER.valueToTree(after);
 
-        JsonNode patchPrevToNew = JsonDiff.asJson(previousJson, newJson);
-        JsonNode patchNewToPrev = JsonDiff.asJson(newJson, previousJson);
+        JsonNode patch = JsonDiff.asJson(beforeNode, afterNode);
+        JsonNode revert = JsonDiff.asJson(afterNode, beforeNode);
 
-        return Pair.of(patchPrevToNew, patchNewToPrev);
+        return Pair.of(patch, revert);
     }
-
-    public static String applyPatches(Object object, List<JsonNode> patches) throws JsonProcessingException {
+    
+    public static JsonNode applyPatchesOnJsonNode(JsonNode jsonObject, List<JsonNode> patches) {
         for (JsonNode patch : patches) {
             try {
                 // Required for ES
-                if (patch.findValue("value") == null) {
-                    ((ObjectNode) patch.get(0)).set("value", (JsonNode) null);
+                if (patch.findValue("value") == null && !patch.isEmpty()) {
+                    ((ObjectNode) patch.get(0)).set("value", null);
                 }
-                JsonNode current = MAPPER.valueToTree(object);
-                object = JsonPatch.fromJson(patch).apply(current);
+                jsonObject = JsonPatch.fromJson(patch).apply(jsonObject);
             } catch (IOException | JsonPatchException e) {
                 throw new RuntimeException(e);
             }
         }
-        return MAPPER.writeValueAsString(object);
+        return jsonObject;
     }
-
-
 }

core/src/main/java/io/kestra/core/server/AbstractServiceLivenessCoordinator.java

@@ -171,7 +171,7 @@ public abstract class AbstractServiceLivenessCoordinator extends AbstractService
     protected void handleAllServiceInNotRunningState() {
         // Soft delete all services which are NOT_RUNNING anymore.
         store.findAllInstancesInStates(Set.of(Service.ServiceState.NOT_RUNNING))
-            .forEach(instance -> safelyUpdate(instance, Service.ServiceState.EMPTY, null));
+            .forEach(instance -> safelyUpdate(instance, Service.ServiceState.INACTIVE, null));
     }
 
     protected void handleAllServicesForTerminatedStates(final Instant now) {

core/src/main/java/io/kestra/core/server/Service.java

@@ -1,9 +1,11 @@
 package io.kestra.core.server;
 
+import com.fasterxml.jackson.annotation.JsonCreator;
 import io.kestra.core.utils.Enums;
 
 import java.util.Arrays;
 import java.util.HashSet;
+import java.util.Map;
 import java.util.Set;
 
 /**
@@ -106,12 +108,12 @@ public interface Service extends AutoCloseable {
      *                         |
      *                         v
      *                  +------+-------+
-     *                  | Empty (8)    |
+     *                  | Inactive (8) |
      *                  +------+-------+
      * </pre>
      */
     enum ServiceState {
-        CREATED(1, 2, 3, 4, 9),            // 0
+        CREATED(1, 2, 3, 4, 9),         // 0
         RUNNING(2, 3, 4, 9),            // 1
         ERROR(4),                       // 2
         DISCONNECTED(4, 7),             // 3
@@ -119,14 +121,24 @@ public interface Service extends AutoCloseable {
         TERMINATED_GRACEFULLY(7),       // 5
         TERMINATED_FORCED(7),           // 6
         NOT_RUNNING(8),                 // 7
-        EMPTY(),                                       // 8 FINAL STATE
-        MAINTENANCE(1, 2, 3, 4);                 // 9
+        INACTIVE(),                                   // 8 FINAL STATE
+        MAINTENANCE(1, 2, 3, 4);        // 9
 
         private final Set<Integer> validTransitions = new HashSet<>();
 
         ServiceState(final Integer... validTransitions) {
             this.validTransitions.addAll(Arrays.asList(validTransitions));
         }
+        
+        @JsonCreator
+        public static ServiceState fromString(final String value) {
+            try {
+                // EMPTY state was renamed to INACTIVE in Kestra 1.0
+                return Enums.getForNameIgnoreCase(value, ServiceState.class, Map.of("EMPTY", INACTIVE));
+            } catch (IllegalArgumentException e) {
+                return INACTIVE;
+            }
+        }
 
         public boolean isValidTransition(final ServiceState newState) {
             return validTransitions.contains(newState.ordinal()) || equals(newState);
@@ -145,7 +157,7 @@ public interface Service extends AutoCloseable {
             return equals(TERMINATED_GRACEFULLY)
                 || equals(TERMINATED_FORCED)
                 || equals(NOT_RUNNING)
-                || equals(EMPTY);
+                || equals(INACTIVE);
         }
 
         public static Set<ServiceState> allRunningStates() {

core/src/main/java/io/kestra/core/server/ServiceLivenessManager.java

@@ -392,7 +392,7 @@ public class ServiceLivenessManager extends AbstractServiceLivenessTask {
                 // More especially, this handles the case where a WORKER is configured with a short gracefulTerminationPeriod
                 // and the JVM was unresponsive for more than this period.
                 // In this context, the worker's tasks have already been resubmitted by the executor; the worker must therefore stop immediately.
-                if (state.equals(Service.ServiceState.NOT_RUNNING) || state.equals(Service.ServiceState.EMPTY)) {
+                if (state.equals(Service.ServiceState.NOT_RUNNING) || state.equals(Service.ServiceState.INACTIVE)) {
                     service.skipGracefulTermination(true);
                 }
                 KestraContext.getContext().shutdown();

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

@@ -130,7 +130,7 @@ public class ConditionService {
         return this.conditionContext(runContext, flow, execution, null);
     }
 
-    boolean valid(FlowInterface flow, List<Condition> list, ConditionContext conditionContext) {
+    public boolean valid(FlowInterface flow, List<Condition> list, ConditionContext conditionContext) {
         return list
             .stream()
             .allMatch(condition -> {

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

@@ -317,6 +317,32 @@ public class ExecutionService {
         return revision != null ? newExecution.withFlowRevision(revision) : newExecution;
     }
 
+    public Execution changeTaskRunState(final Execution execution, Flow flow, String taskRunId, State.Type newState) throws Exception {
+        Execution newExecution = markAs(execution, flow, taskRunId, newState);
+
+        // if the execution was terminated, it could have executed errors/finally/afterExecutions, we must remove them as the execution will be restarted
+        if (execution.getState().isTerminated()) {
+            List<TaskRun> newTaskRuns =  newExecution.getTaskRunList();
+            // We need to remove global error tasks and flowable error tasks if any
+            flow
+                .allErrorsWithChildren()
+                .forEach(task -> newTaskRuns.removeIf(taskRun -> taskRun.getTaskId().equals(task.getId())));
+
+            // We need to remove global finally tasks and flowable error tasks if any
+            flow
+                .allFinallyWithChildren()
+                .forEach(task -> newTaskRuns.removeIf(taskRun -> taskRun.getTaskId().equals(task.getId())));
+
+            // We need to remove afterExecution tasks
+            ListUtils.emptyOnNull(flow.getAfterExecution())
+                .forEach(task -> newTaskRuns.removeIf(taskRun -> taskRun.getTaskId().equals(task.getId())));
+
+            return newExecution.withTaskRunList(newTaskRuns);
+        } else {
+            return newExecution;
+        }
+    }
+
     public Execution markAs(final Execution execution, FlowInterface flow, String taskRunId, State.Type newState) throws Exception {
         return this.markAs(execution, flow, taskRunId, newState, null, null);
     }

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

@@ -3,12 +3,7 @@ package io.kestra.core.services;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import io.kestra.core.exceptions.FlowProcessingException;
 import io.kestra.core.models.executions.Execution;
-import io.kestra.core.models.flows.Flow;
-import io.kestra.core.models.flows.FlowId;
-import io.kestra.core.models.flows.FlowInterface;
-import io.kestra.core.models.flows.FlowWithException;
-import io.kestra.core.models.flows.FlowWithSource;
-import io.kestra.core.models.flows.GenericFlow;
+import io.kestra.core.models.flows.*;
 import io.kestra.core.models.tasks.RunnableTask;
 import io.kestra.core.models.topologies.FlowTopology;
 import io.kestra.core.models.triggers.AbstractTrigger;
@@ -30,16 +25,7 @@ import org.apache.commons.lang3.builder.EqualsBuilder;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.NoSuchElementException;
-import java.util.Objects;
-import java.util.Optional;
+import java.util.*;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
@@ -408,7 +394,7 @@ public class FlowService {
         return latestFlows.values().stream().filter(flow -> !flow.isDeleted());
     }
 
-    protected boolean removeUnwanted(Flow f, Execution execution) {
+    public boolean removeUnwanted(Flow f, Execution execution) {
         // we don't allow recursive
         return !f.uidWithoutRevision().equals(FlowId.uidWithoutRevision(execution));
     }
@@ -551,23 +537,24 @@ public class FlowService {
         return expandAll ? recursiveFlowTopology(new ArrayList<>(), tenant, namespace, id, destinationOnly) : flowTopologyRepository.get().findByFlow(tenant, namespace, id, destinationOnly).stream();
     }
 
-    private Stream<FlowTopology> recursiveFlowTopology(List<FlowId> flowIds, String tenantId, String namespace, String id, boolean destinationOnly) {
+    private Stream<FlowTopology> recursiveFlowTopology(List<String> visitedTopologies, String tenantId, String namespace, String id, boolean destinationOnly) {
         if (flowTopologyRepository.isEmpty()) {
             throw noRepositoryException();
         }
 
-        List<FlowTopology> flowTopologies = flowTopologyRepository.get().findByFlow(tenantId, namespace, id, destinationOnly);
-
-        FlowId flowId = FlowId.of(tenantId, namespace, id, null);
-        if (flowIds.contains(flowId)) {
-            return flowTopologies.stream();
-        }
-        flowIds.add(flowId);
+        var flowTopologies = flowTopologyRepository.get().findByFlow(tenantId, namespace, id, destinationOnly);
 
         return flowTopologies.stream()
-            .flatMap(topology -> Stream.of(topology.getDestination(), topology.getSource()))
-            // recursively fetch child nodes
-            .flatMap(node -> recursiveFlowTopology(flowIds, node.getTenantId(), node.getNamespace(), node.getId(), destinationOnly));
+            // ignore already visited topologies
+            .filter(x -> !visitedTopologies.contains(x.uid()))
+            .flatMap(topology -> {
+                visitedTopologies.add(topology.uid());
+                Stream<FlowTopology> subTopologies = Stream
+                    .of(topology.getDestination(), topology.getSource())
+                    // recursively visit children and parents nodes
+                    .flatMap(relationNode -> recursiveFlowTopology(visitedTopologies, relationNode.getTenantId(), relationNode.getNamespace(), relationNode.getId(), destinationOnly));
+                return Stream.concat(Stream.of(topology), subTopologies);
+            });
     }
 
     private IllegalStateException noRepositoryException() {

core/src/main/java/io/kestra/core/storages/kv/InternalKVStore.java

@@ -10,6 +10,7 @@ import jakarta.annotation.Nullable;
 import java.io.ByteArrayInputStream;
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.net.URI;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.Collections;
@@ -112,7 +113,14 @@ public class InternalKVStore implements KVStore {
     @Override
     public boolean delete(String key) throws IOException {
         KVStore.validateKey(key);
-        return this.storage.delete(this.tenant, this.namespace, this.storageUri(key));
+        URI uri = this.storageUri(key);
+        boolean deleted = this.storage.delete(this.tenant, this.namespace, uri);
+        URI metadataURI = URI.create(uri.getPath() + ".metadata");
+        if (this.storage.exists(this.tenant, this.namespace, metadataURI)){
+            this.storage.delete(this.tenant, this.namespace, metadataURI);
+        }
+        return deleted;
+
     }
 
     /**
@@ -120,18 +128,32 @@ public class InternalKVStore implements KVStore {
      */
     @Override
     public List<KVEntry> list() throws IOException {
-        List<FileAttributes> list;
-        try {
-            list = this.storage.list(this.tenant, this.namespace, this.storageUri(null));
-        } catch (FileNotFoundException e) {
-            return Collections.emptyList();
-        }
+        List<FileAttributes> list = listAllFromStorage();
         return list.stream()
             .map(throwFunction(KVEntry::from))
             .filter(kvEntry -> Optional.ofNullable(kvEntry.expirationDate()).map(expirationDate -> Instant.now().isBefore(expirationDate)).orElse(true))
             .toList();
     }
 
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public List<KVEntry> listAll() throws IOException {
+        List<FileAttributes> list = listAllFromStorage();
+        return list.stream()
+            .map(throwFunction(KVEntry::from))
+            .toList();
+    }
+
+    private List<FileAttributes> listAllFromStorage() throws IOException {
+        try {
+            return this.storage.list(this.tenant, this.namespace, this.storageUri(null));
+        } catch (FileNotFoundException e) {
+            return Collections.emptyList();
+        }
+    }
+
     /**
      * {@inheritDoc}
      */

core/src/main/java/io/kestra/core/storages/kv/KVEntry.java

@@ -1,13 +1,14 @@
 package io.kestra.core.storages.kv;
 
 import io.kestra.core.storages.FileAttributes;
+import jakarta.annotation.Nullable;
 
 import java.io.IOException;
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
 import java.util.Optional;
 
-public record KVEntry(String key, String description, Instant creationDate, Instant updateDate, Instant expirationDate) {
+public record KVEntry(String key, @Nullable String description, Instant creationDate, Instant updateDate, @Nullable Instant expirationDate) {
     public static KVEntry from(FileAttributes fileAttributes) throws IOException {
         Optional<KVMetadata> kvMetadata = Optional.ofNullable(fileAttributes.getMetadata()).map(KVMetadata::new);
         return new KVEntry(

core/src/main/java/io/kestra/core/storages/kv/KVMetadata.java

@@ -1,5 +1,6 @@
 package io.kestra.core.storages.kv;
 
+import lombok.EqualsAndHashCode;
 import lombok.Getter;
 
 import java.time.Duration;
@@ -9,6 +10,7 @@ import java.util.Map;
 import java.util.Optional;
 
 @Getter
+@EqualsAndHashCode
 public class KVMetadata {
     private String description;
     private Instant expirationDate;
@@ -17,14 +19,18 @@ public class KVMetadata {
         if (ttl != null && ttl.isNegative()) {
             throw new IllegalArgumentException("ttl cannot be negative");
         }
-
-
+        
         this.description = description;
         if (ttl != null) {
             this.expirationDate = Instant.now().plus(ttl);
         }
     }
-
+    
+    public KVMetadata(String description, Instant expirationDate) {
+        this.description = description;
+        this.expirationDate = expirationDate;
+    }
+    
     public KVMetadata(Map<String, String> metadata) {
         if (metadata == null) {
             return;
@@ -46,4 +52,9 @@ public class KVMetadata {
         }
         return map;
     }
+    
+    @Override
+    public String toString() {
+        return "[description=" + description + ", expirationDate=" + expirationDate + "]";
+    }
 }

core/src/main/java/io/kestra/core/storages/kv/KVStore.java

@@ -1,12 +1,12 @@
 package io.kestra.core.storages.kv;
 
 import io.kestra.core.exceptions.ResourceExpiredException;
-import io.kestra.core.runners.RunContext;
 import io.kestra.core.storages.StorageContext;
-import jakarta.annotation.Nullable;
 
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.net.URI;
+import java.time.Duration;
 import java.util.List;
 import java.util.Optional;
 import java.util.regex.Pattern;
@@ -80,6 +80,14 @@ public interface KVStore {
      */
     List<KVEntry> list() throws IOException;
 
+    /**
+     * Lists all the K/V store entries, expired or not.
+     *
+     * @return  The list of all {@link KVEntry}.
+     * @throws IOException if an error occurred while executing the operation on the K/V store.
+     */
+    List<KVEntry> listAll() throws IOException;
+
     /**
      * Finds the K/V store entry for the given key.
      *
@@ -98,8 +106,33 @@ public interface KVStore {
     default boolean exists(String key) throws IOException {
         return list().stream().anyMatch(kvEntry -> kvEntry.key().equals(key));
     }
-
-
+    
+    /**
+     * Finds a KV entry with associated metadata for a given key.
+     *
+     * @param key   the KV entry key.
+     * @return an optional of {@link KVValueAndMetadata}.
+     * 
+     * @throws UncheckedIOException if an error occurred while executing the operation on the K/V store.
+     */
+    default Optional<KVValueAndMetadata> findMetadataAndValue(final String key) throws UncheckedIOException {
+        try {
+            return get(key).flatMap(entry ->
+                {
+                    try {
+                        return getValue(entry.key()).map(current -> new KVValueAndMetadata(new KVMetadata(entry.description(), entry.expirationDate()), current.value()));
+                    } catch (IOException e) {
+                        throw new UncheckedIOException(e);
+                    } catch (ResourceExpiredException e) {
+                        return Optional.empty();
+                    }
+                }
+            );
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
+    }
+    
     Pattern KEY_VALIDATOR_PATTERN = Pattern.compile("[a-zA-Z0-9][a-zA-Z0-9._-]*");
 
     /**

core/src/main/java/io/kestra/core/topologies/FlowTopologyService.java

@@ -18,6 +18,7 @@ import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.repositories.FlowTopologyRepositoryInterface;
 import io.kestra.core.services.ConditionService;
 import io.kestra.core.utils.ListUtils;
+import io.kestra.core.utils.MapUtils;
 import io.kestra.plugin.core.condition.*;
 import io.micronaut.core.annotation.Nullable;
 import jakarta.inject.Inject;
@@ -175,9 +176,6 @@ public class FlowTopologyService {
     protected boolean isTriggerChild(Flow parent, Flow child) {
         List<AbstractTrigger> triggers = ListUtils.emptyOnNull(child.getTriggers());
 
-        // simulated execution: we add a "simulated" label so conditions can know that the evaluation is for a simulated execution
-        Execution execution = Execution.newExecution(parent, (f, e) -> null, List.of(SIMULATED_EXECUTION), Optional.empty());
-
         // keep only flow trigger
         List<io.kestra.plugin.core.trigger.Flow> flowTriggers = triggers
             .stream()
@@ -189,13 +187,16 @@ public class FlowTopologyService {
             return false;
         }
 
+        // simulated execution: we add a "simulated" label so conditions can know that the evaluation is for a simulated execution
+        Execution execution = Execution.newExecution(parent, (f, e) -> null, List.of(SIMULATED_EXECUTION), Optional.empty());
+
         boolean conditionMatch =  flowTriggers
             .stream()
             .flatMap(flow -> ListUtils.emptyOnNull(flow.getConditions()).stream())
             .allMatch(condition -> validateCondition(condition, parent, execution));
 
         boolean preconditionMatch = flowTriggers.stream()
-            .anyMatch(flow -> flow.getPreconditions() == null || validateMultipleConditions(flow.getPreconditions().getConditions(), parent, execution));
+            .anyMatch(flow -> flow.getPreconditions() == null || validatePreconditions(flow.getPreconditions(), parent, execution));
 
         return conditionMatch && preconditionMatch;
     }
@@ -209,7 +210,13 @@ public class FlowTopologyService {
             return validateMultipleConditions(multipleCondition.getConditions(), child, execution);
         }
 
-        return this.conditionService.isValid(condition, child, execution);
+        try {
+            return this.conditionService.isValid(condition, child, execution);
+        } catch (Exception e) {
+            // extra safety net, it means there is a bug
+            log.error("unable to validate condition in FlowTopologyService, flow: {}, condition: {}", child.uid(), condition, e);
+            return false;
+        }
     }
 
     private boolean validateMultipleConditions(Map<String, Condition> multipleConditions, FlowInterface child, Execution execution) {
@@ -233,11 +240,24 @@ public class FlowTopologyService {
     }
 
     private boolean isMandatoryMultipleCondition(Condition condition) {
-        return Stream
-            .of(
-                Expression.class
-            )
-            .anyMatch(aClass -> condition.getClass().isAssignableFrom(aClass));
+        return condition.getClass().isAssignableFrom(Expression.class);
+    }
+
+    private boolean validatePreconditions(io.kestra.plugin.core.trigger.Flow.Preconditions preconditions, FlowInterface child, Execution execution) {
+        boolean  upstreamFlowMatched = MapUtils.emptyOnNull(preconditions.getUpstreamFlowsConditions())
+            .values()
+            .stream()
+            .filter(c -> !isFilterCondition(c))
+            .anyMatch(c -> validateCondition(c, child, execution));
+
+        boolean  whereMatched = MapUtils.emptyOnNull(preconditions.getWhereConditions())
+            .values()
+            .stream()
+            .filter(c -> !isFilterCondition(c))
+            .allMatch(c -> validateCondition(c, child, execution));
+
+        // to be a dependency, if upstream flow is set it must be either inside it so it's a AND between upstream flow and where
+        return upstreamFlowMatched && whereMatched;
     }
 
     private boolean isFilterCondition(Condition condition) {

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

@@ -0,0 +1,15 @@
+package io.kestra.core.utils;
+
+import jakarta.inject.Singleton;
+
+@Singleton
+public class EditionProvider {
+    public Edition get() {
+        return Edition.OSS;
+    }
+
+    public enum Edition {
+        OSS,
+        EE
+    }
+}

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

@@ -206,22 +206,17 @@ public class MapUtils {
 
     /**
      * Utility method that flatten a nested map.
-     * <p>
-     * NOTE: for simplicity, this method didn't allow to flatten maps with conflicting keys that would end up in different flatten keys,
-     * this could be related later if needed by flattening {k1: k2: {k3: v1}, k1: {k4: v2}} to {k1.k2.k3: v1, k1.k4: v2} is prohibited for now.
      *
      * @param nestedMap the nested map.
      * @return the flattened map.
-     *
-     * @throws IllegalArgumentException if any entry contains a map of more than one element.
      */
     public static Map<String, Object> nestedToFlattenMap(@NotNull Map<String, Object> nestedMap) {
         Map<String, Object> result = new TreeMap<>();
 
         for (Map.Entry<String, Object> entry : nestedMap.entrySet()) {
             if (entry.getValue() instanceof Map<?, ?> map) {
-                Map.Entry<String, Object> flatten = flattenEntry(entry.getKey(), (Map<String, Object>) map);
-                result.put(flatten.getKey(), flatten.getValue());
+                Map<String, Object> flatten = flattenEntry(entry.getKey(), (Map<String, Object>) map);
+                result.putAll(flatten);
             } else {
                 result.put(entry.getKey(), entry.getValue());
             }
@@ -229,18 +224,19 @@ public class MapUtils {
         return result;
     }
 
-    private static Map.Entry<String, Object> flattenEntry(String key, Map<String, Object> value) {
-        if (value.size() > 1) {
-            throw new IllegalArgumentException("You cannot flatten a map with an entry that is a map of more than one element, conflicting key: " + key);
+    private static Map<String, Object> flattenEntry(String key, Map<String, Object> value) {
+        Map<String, Object> result = new TreeMap<>();
+
+        for (Map.Entry<String, Object> entry : value.entrySet()) {
+            String newKey = key + "." + entry.getKey();
+            Object newValue = entry.getValue();
+            if (newValue instanceof Map<?, ?> map) {
+                result.putAll(flattenEntry(newKey, (Map<String, Object>) map));
+            } else {
+                result.put(newKey, newValue);
+            }
         }
 
-        Map.Entry<String, Object> entry = value.entrySet().iterator().next();
-        String newKey = key + "." + entry.getKey();
-        Object newValue = entry.getValue();
-        if (newValue instanceof Map<?, ?> map) {
-            return flattenEntry(newKey, (Map<String, Object>) map);
-        } else {
-            return Map.entry(newKey, newValue);
-        }
+        return result;
     }
 }

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

@@ -36,6 +36,19 @@ import static io.kestra.core.models.Label.SYSTEM_PREFIX;
 @Singleton
 @Introspected
 public class FlowValidator implements ConstraintValidator<FlowValidation, Flow> {
+    public static List<String> RESERVED_FLOW_IDS = List.of(
+        "pause",
+        "resume",
+        "force-run",
+        "change-status",
+        "kill",
+        "executions",
+        "search",
+        "source",
+        "disable",
+        "enable"
+    );
+
     @Inject
     private FlowService flowService;
 
@@ -50,6 +63,10 @@ public class FlowValidator implements ConstraintValidator<FlowValidation, Flow>
 
         List<String> violations = new ArrayList<>();
 
+        if (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));
+        }
+
         if (flowService.requireExistingNamespace(value.getTenantId(), value.getNamespace())) {
             violations.add("Namespace '" + value.getNamespace() + "' does not exist but is required to exist before a flow can be created in it.");
         }

core/src/main/java/io/kestra/plugin/core/condition/ExecutionLabels.java

@@ -29,7 +29,7 @@ import java.util.Map;
 @Getter
 @NoArgsConstructor
 @Schema(
-    title = "Condition that check labels of an execution."
+    title = "Condition that checks labels of an execution."
 )
 @Plugin(
     examples = {

core/src/main/java/io/kestra/plugin/core/condition/ExecutionOutputs.java

@@ -35,8 +35,8 @@ import static io.kestra.core.utils.MapUtils.mergeWithNullableValues;
     examples = {
         @Example(
             title = """
-                The upstream `flow_a` must explicitly define its outputs 
-                to be used in the `ExecutionOutputs` condition. 
+                The upstream `flow_a` must explicitly define its outputs
+                to be used in the `ExecutionOutputs` condition.
 
                 ```yaml
                 id: flow_a
@@ -58,7 +58,7 @@ import static io.kestra.core.utils.MapUtils.mergeWithNullableValues;
                     value: "{{ outputs.hello.value }}"
                 ```
 
-                The `flow_condition_executionoutputs` will run whenever `flow_a` finishes successfully 
+                The `flow_condition_executionoutputs` will run whenever `flow_a` finishes successfully
                 and returns an output matching the value 'hello':
                 """,
             full = true,
@@ -105,7 +105,7 @@ public class ExecutionOutputs extends Condition implements ScheduleCondition {
             conditionContext.getVariables(),
             Map.of(TRIGGER_VAR, Map.of(OUTPUTS_VAR, conditionContext.getExecution().getOutputs()))
         );
-        
+
         return conditionContext.getRunContext().render(expression).skipCache().as(Boolean.class, variables).orElseThrow();
     }
 

core/src/main/java/io/kestra/plugin/core/condition/Expression.java

@@ -35,7 +35,7 @@ import jakarta.validation.constraints.NotNull;
             code = """
                 id: myflow
                 namespace: company.team
-                
+
                 tasks:
                   - id: hello
                     type: io.kestra.plugin.core.log.Log

core/src/main/java/io/kestra/plugin/core/condition/FlowNamespaceCondition.java

@@ -47,7 +47,7 @@ public class FlowNamespaceCondition extends Condition {
 
     @Builder.Default
     @Schema(
-        title = "If we must look at the flow namespace by prefix (checked using startWith). The prefix is case sensitive."
+        title = "If we must look at the flow namespace by prefix (checked using startsWith). The prefix is case sensitive."
     )
     @PluginProperty
     private final Boolean prefix = false;

core/src/main/java/io/kestra/plugin/core/condition/MultipleCondition.java

@@ -26,7 +26,7 @@ import java.util.*;
 @Getter
 @NoArgsConstructor
 @Schema(
-    title = "Run a flow if the list of preconditions are met in a time window.",
+    title = "Run a flow if the list of preconditions is met in a time window.",
     description = """
         **This task is deprecated**, use the `preconditions` property of the `io.kestra.plugin.core.trigger.Flow` trigger instead.
         Will trigger an executions when all the flows defined by the preconditions are successfully executed in a specific period of time.

core/src/main/java/io/kestra/plugin/core/condition/Not.java

@@ -27,7 +27,7 @@ import static io.kestra.core.utils.Rethrow.throwPredicate;
 @Getter
 @NoArgsConstructor
 @Schema(
-    title = "Condition to exclude others conditions."
+    title = "Condition to exclude other conditions."
 )
 @Plugin(
     examples = {
@@ -62,7 +62,7 @@ public class Not extends Condition implements ScheduleCondition {
     @NotEmpty
     @Schema(
         title = "The list of conditions to exclude.",
-        description = "If any conditions is true, it will prevent the event's execution."
+        description = "If any condition is true, it will prevent the event's execution."
     )
     @PluginProperty
     private List<Condition> conditions;

core/src/main/java/io/kestra/plugin/core/condition/Or.java

@@ -31,7 +31,7 @@ import static io.kestra.core.utils.Rethrow.throwPredicate;
 @Plugin(
     examples = {
         @Example(
-            title = "Trigger condition to execute the flow when any of the condition is satisfied.",
+            title = "Trigger condition to execute the flow when any of the conditions are satisfied.",
             full = true,
             code = """
                 id: schedule_condition_or

core/src/main/java/io/kestra/plugin/core/condition/PublicHoliday.java

@@ -5,14 +5,12 @@ import de.focus_shift.jollyday.core.ManagerParameters;
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.models.annotations.Example;
 import io.kestra.core.models.annotations.Plugin;
-import io.kestra.core.models.annotations.PluginProperty;
 import io.kestra.core.models.conditions.Condition;
 import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.conditions.ScheduleCondition;
 import io.kestra.core.models.property.Property;
 import io.kestra.core.utils.DateUtils;
 import io.swagger.v3.oas.annotations.media.Schema;
-import jakarta.validation.constraints.NotEmpty;
 import jakarta.validation.constraints.NotNull;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
@@ -31,7 +29,7 @@ import java.time.LocalDate;
     examples = {
         @Example(
             full = true,
-            title = "Trigger condition to excute the flow only on public holidays.",
+            title = "Trigger condition to execute the flow only on public holidays.",
             code = """
                 id: schedule_condition_public-holiday
                 namespace: company.team
@@ -52,7 +50,7 @@ import java.time.LocalDate;
         ),
         @Example(
             full = true,
-            title = "Trigger condition to excute the flow only on work days in France.",
+            title = "Trigger condition to execute the flow only on work days in France.",
             code = """
                 id: schedule-condition-work-days
                 namespace: company.team

core/src/main/java/io/kestra/plugin/core/dashboard/chart/Bar.java

@@ -25,7 +25,7 @@ import lombok.experimental.SuperBuilder;
 @Plugin(
     examples = {
         @Example(
-            title = "Display a bar chart with with Executions per Namespace.",
+            title = "Display a bar chart with Executions per Namespace.",
             full = true,
             code = { """
                 charts:

core/src/main/java/io/kestra/plugin/core/dashboard/chart/Markdown.java

@@ -42,7 +42,6 @@ import lombok.experimental.SuperBuilder;
 
                       - Use this insight to identify trends and optimize performance.
                 """
-                
             }
         )
     }

core/src/main/java/io/kestra/plugin/core/dashboard/chart/Pie.java

@@ -25,7 +25,7 @@ import lombok.experimental.SuperBuilder;
 @Plugin(
     examples = {
         @Example(
-            title = "Display a pie chart with with Executions per State.",
+            title = "Display a pie chart with Executions per State.",
             full = true,
             code = { """
                 charts:

core/src/main/java/io/kestra/plugin/core/dashboard/chart/Table.java

@@ -25,7 +25,7 @@ import lombok.experimental.SuperBuilder;
 @Plugin(
     examples = {
         @Example(
-            title = "Display a table with a Log count for each level by Namespace.",
+            title = "Display a table with Log counts for each level by Namespace.",
             full = true,
             code = { """
                 charts:

core/src/main/java/io/kestra/plugin/core/dashboard/chart/TimeSeries.java

@@ -29,7 +29,7 @@ import lombok.experimental.SuperBuilder;
         @Example(
             title = "Display a chart with Executions over the last week.",
             full = true,
-            code = { """    
+            code = { """
                 charts:
                   - id: executions_timeseries
                     type: io.kestra.plugin.core.dashboard.chart.TimeSeries

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

@@ -31,7 +31,7 @@ import lombok.experimental.SuperBuilder;
             title = "Display a chart with a Executions per Namespace broken out by State.",
             full = true,
             code = { """
-            charts:                
+            charts:
               - id: executions_per_namespace_bars
                 type: io.kestra.plugin.core.dashboard.chart.Bar
                 chartOptions:

core/src/main/java/io/kestra/plugin/core/debug/Return.java

@@ -50,7 +50,7 @@ import java.util.Optional;
                     Bearer {{ inputs.token }}
                     {%- elseif inputs.username is not empty and inputs.password is not empty -%}
                     Basic {{ (inputs.username + ':' + inputs.password) | base64encode }}
-                    {%- endif -%} 
+                    {%- endif -%}
                 """
         )
     },

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

@@ -73,7 +73,7 @@ public class Assert extends Task implements RunnableTask<VoidOutput> {
     @PluginProperty(dynamic = true)
     private List<String> conditions;
 
-    @Schema(title = "Optional error message.")
+    @Schema(title = "Optional error message")
     private Property<String> errorMessage;
 
     @Override

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

@@ -37,7 +37,7 @@ import static io.kestra.core.utils.Rethrow.throwPredicate;
 @Plugin(
     examples = {
         @Example(
-            title = "Send a slack notification if there is no execution for a flow for the last 24 hours.",
+            title = "Send a slack notification if there is no execution for a flow in the last 24 hours.",
             full = true,
             code = """
                 id: executions_count
@@ -78,32 +78,32 @@ import static io.kestra.core.utils.Rethrow.throwPredicate;
 )
 public class Count extends Task implements RunnableTask<Count.Output> {
     @Schema(
-        title = "A list of flows to be filtered.",
+        title = "A list of flows to be filtered",
         description = "If not provided, namespaces must be set."
     )
     @PluginProperty
     protected List<Flow> flows;
 
     @Schema(
-        title = "A list of states to be filtered."
+        title = "A list of states to be filtered"
     )
     protected Property<List<State.Type>> states;
 
     @NotNull
     @Schema(
-        title = "The start date."
+        title = "The start date"
     )
     protected Property<String> startDate;
 
     @Schema(
-        title = "The end date."
+        title = "The end date"
     )
     protected Property<String> endDate;
 
     @NotNull
     @Schema(
-        title = "The expression to look at against each flow.",
-        description = "The expression is such that expression must return `true` in order to keep the current line.\n" +
+        title = "The expression to check against each flow",
+        description = "The expression is such that the expression must return `true` in order to keep the current line.\n" +
             "Some examples: \n" +
             "- ```yaml {{ eq count 0 }} ```: no execution found\n" +
             "- ```yaml {{ gte count 5 }} ```: more than 5 executions\n"
@@ -121,7 +121,7 @@ public class Count extends Task implements RunnableTask<Count.Output> {
             .getBean(ExecutionRepositoryInterface.class);
 
         if (flows == null && namespaces == null) {
-            throw new IllegalArgumentException("You must provide a list of flows or namespaces");
+            throw new IllegalArgumentException("You must provide a list of flows or namespaces.");
         }
 
         var flowInfo = runContext.flowInfo();

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

@@ -21,6 +21,7 @@ import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.constraints.NotNull;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
+import lombok.extern.slf4j.Slf4j;
 
 import java.util.Optional;
 
@@ -68,6 +69,7 @@ import java.util.Optional;
         )
     }
 )
+@Slf4j
 public class Exit extends Task implements ExecutionUpdatableTask {
     @NotNull
     @Schema(
@@ -104,12 +106,13 @@ public class Exit extends Task implements ExecutionUpdatableTask {
                     // ends all parents
                     while (newTaskRun.getParentTaskRunId() != null) {
                         newTaskRun = newExecution.findTaskRunByTaskRunId(newTaskRun.getParentTaskRunId()).withState(exitState);
-                        newExecution = execution.withTaskRun(newTaskRun);
+                        newExecution = newExecution.withTaskRun(newTaskRun);
                     }
                     return newExecution;
                 } catch (InternalException e) {
                     // in case we cannot update the last not terminated task run, we ignore it
-                    return execution;
+                    log.warn("Unable to update the taskrun state", e);
+                    return execution.withState(exitState);
                 }
             })
             .orElse(execution)

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

@@ -119,7 +119,7 @@ public class Fail extends Task implements RunnableTask<VoidOutput> {
     )
     private Property<String> condition;
 
-    @Schema(title = "Optional error message.")
+    @Schema(title = "Optional error message")
     @Builder.Default
     private Property<String> errorMessage = Property.ofValue("Task failure");
 

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

@@ -36,7 +36,7 @@ import static io.kestra.core.utils.Rethrow.throwFunction;
 @Getter
 @NoArgsConstructor
 @Schema(
-    title = "Allow to add or overwrite labels for the current execution at runtime.",
+    title = "Add or overwrite labels for the current execution at runtime.",
     description = "Trying to pass a system label (a label starting with `system.`) will fail the task."
 )
 @Plugin(
@@ -76,7 +76,7 @@ public class Labels extends Task implements ExecutionUpdatableTask {
     private static final ObjectMapper MAPPER = JacksonMapper.ofJson();
 
     @Schema(
-        title = "Labels to add to the current execution.",
+        title = "Labels to add to the current execution",
         description = "The value should result in a list of labels or a labelKey:labelValue map",
         oneOf = {
             String.class,

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

@@ -45,59 +45,59 @@ import java.util.List;
 )
 public class PurgeExecutions extends Task implements RunnableTask<PurgeExecutions.Output> {
     @Schema(
-        title = "Namespace whose flows need to be purged, or namespace of the flow that needs to be purged.",
+        title = "Namespace whose flows need to be purged, or namespace of the flow that needs to be purged",
         description = "If `flowId` isn't provided, this is a namespace prefix, else the namespace of the flow."
     )
     private Property<String> namespace;
 
     @Schema(
-        title = "The flow ID to be purged.",
+        title = "The flow ID to be purged",
         description = "You need to provide the `namespace` properties if you want to purge a flow."
     )
     private Property<String> flowId;
 
     @Schema(
-        title = "The minimum date to be purged.",
+        title = "The date after which data should be purged",
         description = "All data of flows executed after this date will be purged."
     )
     private Property<String> startDate;
 
     @Schema(
-        title = "The maximum date to be purged.",
+        title = "The date before which data should be purged.",
         description = "All data of flows executed before this date will be purged."
     )
     @NotNull
     private Property<String> endDate;
 
     @Schema(
-        title = "The state of the executions to be purged.",
+        title = "The state of the executions to be purged",
         description = "If not set, executions for any states will be purged."
     )
     private Property<List<State.Type>> states;
 
     @Schema(
-        title = "Whether to purge executions."
+        title = "Flag specifying whether to purge executions"
     )
     @Builder.Default
     private Property<Boolean> purgeExecution = Property.ofValue(true);
 
     @Schema(
-        title = "Whether to purge execution's logs.",
+        title = "Flag specifying whether to purge execution logs",
         description = """
-            This will only purge logs from executions not from triggers, and it will do it execution by execution.
-            The `io.kestra.plugin.core.log.PurgeLogs` task is a better fit to purge logs as it will purge logs in bulk, and will also purge logs not tied to an execution like trigger logs."""
+            This will only purge logs from executions, not from triggers, and it will do it execution by execution.
+            The `io.kestra.plugin.core.log.PurgeLogs` task is a better fit to purge, as it will purge logs in bulk and will also purge logs not tied to an execution like trigger logs."""
     )
     @Builder.Default
     private Property<Boolean> purgeLog = Property.ofValue(true);
 
     @Schema(
-        title = "Whether to purge execution's metrics."
+        title = "Flag specifying whether to purge execution's metrics."
     )
     @Builder.Default
     private Property<Boolean> purgeMetric = Property.ofValue(true);
 
     @Schema(
-        title = "Whether to purge execution's files from the Kestra's internal storage."
+        title = "Flag specifying whether to purge execution's files from the Kestra's internal storage"
     )
     @Builder.Default
     private Property<Boolean> purgeStorage = Property.ofValue(true);
@@ -141,22 +141,22 @@ public class PurgeExecutions extends Task implements RunnableTask<PurgeExecution
     @Getter
     public static class Output implements io.kestra.core.models.tasks.Output {
         @Schema(
-            title = "The count of deleted executions."
+            title = "The count of deleted executions"
         )
         private int executionsCount;
 
         @Schema(
-            title = "The count of deleted logs."
+            title = "The count of deleted logs"
         )
         private int logsCount;
 
         @Schema(
-            title = "The count of deleted storage files."
+            title = "The count of deleted storage files"
         )
         private int storagesCount;
 
         @Schema(
-            title = "The count of deleted metrics."
+            title = "The count of deleted metrics"
         )
         private int metricsCount;
     }

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

@@ -64,14 +64,14 @@ public class Resume  extends Task implements RunnableTask<VoidOutput> {
         description = """
             If you explicitly define an `executionId`, Kestra will use that specific ID.
 
-            If another `namespace` and `flowId` properties are set, Kestra will look for a paused execution for that corresponding flow.
+            If `executionId` is not set and `namespace` and `flowId` properties are set, Kestra will look for a paused execution for that corresponding flow.
 
             If `executionId` is not set, the task will use the ID of the current execution."""
     )
     private Property<String> executionId;
 
     @Schema(
-        title = "Inputs to be passed to the execution when it's resumed."
+        title = "Inputs to be passed to the execution when it's resumed"
     )
     private Property<Map<String, Object>> inputs;
 
@@ -93,7 +93,7 @@ public class Resume  extends Task implements RunnableTask<VoidOutput> {
 
         Execution execution = executionRepository.findById(executionInfo.tenantId(), executionInfo.id())
             .orElseThrow(() -> new IllegalArgumentException("No execution found for execution id " + executionInfo.id()));
-        FlowInterface flow = flowExecutor.findByExecution(execution).orElseThrow(() -> new IllegalArgumentException("Flow not found for execution id " + executionInfo.id()));
+        FlowInterface flow = flowExecutor.findByExecution(execution).orElseThrow(() -> new IllegalArgumentException("Flow not found for execution ID " + executionInfo.id()));
 
         Map<String, Object> renderedInputs = runContext.render(this.inputs).asMap(String.class, Object.class);
         renderedInputs = !renderedInputs.isEmpty() ? renderedInputs : null;

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

@@ -22,7 +22,7 @@ import java.util.Map;
 @Getter
 @NoArgsConstructor
 @Schema(
-    title = "Allow to set execution variables. These variables will then be available via the `{{ vars.name }}` expression."
+    title = "Allow to set execution variables. These variables are available via the `{{ vars.name }}` expression."
 )
 @Plugin(
     examples = {
@@ -53,7 +53,7 @@ public class SetVariables extends Task implements ExecutionUpdatableTask {
     @NotNull
     private Property<Map<String, Object>> variables;
 
-    @Schema(title = "Whether to overwrite existing variables")
+    @Schema(title = "Flag specifying whether to overwrite existing variables")
     @NotNull
     @Builder.Default
     private Property<Boolean> overwrite = Property.ofValue(true);

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

@@ -21,7 +21,7 @@ import java.util.Map;
 @Getter
 @NoArgsConstructor
 @Schema(
-    title = "Allow to unset execution variables."
+    title = "Unset execution variables."
 )
 @Plugin(
     examples = {
@@ -57,7 +57,7 @@ public class UnsetVariables extends Task implements ExecutionUpdatableTask {
     @NotNull
     private Property<List<String>> variables;
 
-    @Schema(title = "Whether to ignore missing variables")
+    @Schema(title = "Flag specifying whether to ignore missing variables")
     @NotNull
     @Builder.Default
     private Property<Boolean> ignoreMissing = Property.ofValue(false);

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

@@ -91,7 +91,7 @@ public class Dag extends Task implements FlowableTask<VoidOutput> {
     @NotNull
     @Builder.Default
     @Schema(
-        title = "Number of concurrent parallel tasks that can be running at any point in time.",
+        title = "Number of concurrent parallel tasks that can be running at any point in time",
         description = "If the value is `0`, no concurrency limit exists for the tasks in a DAG and all tasks that can run in parallel will start at the same time."
     )
     private final Property<Integer> concurrent = Property.ofValue(0);
@@ -134,7 +134,7 @@ public class Dag extends Task implements FlowableTask<VoidOutput> {
     private void controlTask() throws IllegalVariableEvaluationException {
         List<String> dagCheckNotExistTasks = this.dagCheckNotExistTask(this.tasks);
         if (!dagCheckNotExistTasks.isEmpty()) {
-            throw new IllegalVariableEvaluationException("Some task doesn't exists on task '" + this.id + "': " +  String.join(", ", dagCheckNotExistTasks));
+            throw new IllegalVariableEvaluationException("Some task doesn't exist on task '" + this.id + "': " +  String.join(", ", dagCheckNotExistTasks));
         }
 
         ArrayList<String> cyclicDependenciesTasks = this.dagCheckCyclicDependencies(this.tasks);
@@ -238,14 +238,14 @@ public class Dag extends Task implements FlowableTask<VoidOutput> {
     public static class DagTask {
         @NotNull
         @Schema(
-            title = "The task within the DAG."
+            title = "The task within the DAG"
         )
         @PluginProperty
         private Task task;
 
         @PluginProperty
         @Schema(
-            title = "The list of task IDs that should have been successfully executed before starting this task."
+            title = "The list of task IDs that should have been successfully executed before starting this task"
         )
         private List<String> dependsOn;
     }

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

@@ -126,15 +126,15 @@ public class EachParallel extends Parallel implements FlowableTask<VoidOutput> {
     @NotNull
     @Builder.Default
     @Schema(
-        title = "Number of concurrent parallel tasks that can be running at any point in time.",
-        description = "If the value is `0`, no limit exist and all the tasks will start at the same time."
+        title = "Number of concurrent parallel tasks that can be running at any point in time",
+        description = "If the value is `0`, no limit exists and all the tasks will start at the same time."
     )
     private final Property<Integer> concurrent = Property.ofValue(0);
 
     @NotNull
     @PluginProperty(dynamic = true)
     @Schema(
-        title = "The list of values for this task.",
+        title = "The list of values for this task",
         description = "The value can be passed as a string, a list of strings, or a list of objects.",
         oneOf = {String.class, Object[].class}
     )