NiFi Process Group Restart Recovery Gate with Kestra

id: nifi-process-group-restart-recovery-gate
namespace: company.team

description: |
  Start a previously stopped Apache NiFi Process Group only when explicitly confirmed, re-read
  its status right after, and alert Slack when the restart did not actually bring any component
  back to active, instead of trusting the Start API call's 200 response alone. Logs a governance
  audit trail either way. Companion to this repo's NiFi backlog and stall sentinel, which
  documents starting a group back up as a deliberate, separate flow run.

inputs:
  - id: nifi_url
    type: STRING
    displayName: NiFi base URL
    description: Fully qualified URL of the NiFi cluster or instance. Maps to every
      task's url field.
    defaults: "https://localhost:8443"

  - id: nifi_username
    type: STRING
    displayName: NiFi username
    description: Username for NiFi's single-user JWT authentication. Maps to every
      task's username field.
    defaults: "admin"

  - id: ssl_verify
    type: BOOL
    displayName: Verify SSL certificate
    description: Set false for a local instance with a self-signed certificate, true
      for a production instance with a valid certificate.
    defaults: false

  - id: process_group_id
    type: STRING
    displayName: Process Group ID
    description: NiFi Process Group UUID to restart, or the literal value root for
      the root Process Group.
    defaults: "root"

  - id: confirm_start
    type: BOOL
    displayName: Confirm start
    description: Must be true for start_process_group to actually run. When false,
      this flow only probes current status and logs what it would do, nothing in
      NiFi is changed.
    defaults: false

  - id: min_active_threads_after_start
    type: INT
    displayName: Min active threads after start
    description: After starting, the group's activeThreadCount must reach at least
      this many for the restart to be classified RECOVERY_CONFIRMED.
    defaults: 1

tasks:
  - id: probe_before
    type: io.kestra.plugin.nifi.GetProcessGroupStatus
    description: Reads queuedCount and activeThreadCount before any action, for the
      audit log and the dry-run branch.
    allowFailure: true
    url: "{{ inputs.nifi_url }}"
    username: "{{ inputs.nifi_username }}"
    password: "{{ secret('NIFI_PASSWORD') }}"
    sslVerify: "{{ inputs.ssl_verify }}"
    processGroupId: "{{ inputs.process_group_id }}"

  - id: route_confirm
    type: io.kestra.plugin.core.flow.If
    description: Only start the group when confirm_start is explicitly true. False
      is always safe, nothing in NiFi is touched.
    condition: "{{ inputs.confirm_start }}"
    then:
      - id: start_process_group
        type: io.kestra.plugin.nifi.StartProcessGroup
        description: Schedules every component in this one Process Group to RUNNING.
          Does not itself confirm anything actually started processing, that is
          checked by probe_after below.
        allowFailure: true
        url: "{{ inputs.nifi_url }}"
        username: "{{ inputs.nifi_username }}"
        password: "{{ secret('NIFI_PASSWORD') }}"
        sslVerify: "{{ inputs.ssl_verify }}"
        processGroupId: "{{ inputs.process_group_id }}"
    else:
      - id: log_dry_run
        type: io.kestra.plugin.core.log.Log
        message: "confirm_start is false, nothing started. Current state for {{
          inputs.process_group_id }}: queuedCount={{
          outputs.probe_before.queuedCount }}, activeThreadCount={{
          outputs.probe_before.activeThreadCount }}."

  - id: probe_after
    type: io.kestra.plugin.nifi.GetProcessGroupStatus
    description: Re-reads status immediately after the branch above, whether or not
      a start actually ran, so evaluate always has a current reading to classify
      against.
    allowFailure: true
    url: "{{ inputs.nifi_url }}"
    username: "{{ inputs.nifi_username }}"
    password: "{{ secret('NIFI_PASSWORD') }}"
    sslVerify: "{{ inputs.ssl_verify }}"
    processGroupId: "{{ inputs.process_group_id }}"

  - id: evaluate
    type: io.kestra.plugin.core.output.OutputValues
    description: Classifies the restart outcome from probe_after, not from whether
      the Start API call merely returned success.
    values:
      severity: >-
        {{ inputs.confirm_start == false ? 'DRY_RUN' : (outputs.probe_after is
        not defined ? 'CRITICAL' : ((outputs.probe_after.activeThreadCount is
        defined ? outputs.probe_after.activeThreadCount : 0) >=
        inputs.min_active_threads_after_start ? 'RECOVERY_CONFIRMED' :
        'RECOVERY_NOT_CONFIRMED')) }}

  - id: route_result
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.evaluate.values.severity == 'CRITICAL' or
      outputs.evaluate.values.severity == 'RECOVERY_NOT_CONFIRMED' }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': outputs.evaluate.values.severity ~ ': NiFi Process Group ' ~ inputs.process_group_id ~ ' restart did not confirm. queuedCount=' ~ (outputs.probe_after.queuedCount ?? 'unknown') ~ ', activeThreadCount=' ~ (outputs.probe_after.activeThreadCount ?? 'unknown') ~ '. The operator fix may not have resolved the stuck component, or NiFi has not finished spinning threads up yet, rerun this flow in a minute to rule that out.'} | toJson }}
    else:
      - id: compliant
        type: io.kestra.plugin.core.log.Log
        message: "NiFi Process Group {{ inputs.process_group_id }}: {{
          outputs.evaluate.values.severity }}. queuedCount={{
          outputs.probe_after.queuedCount }}, activeThreadCount={{
          outputs.probe_after.activeThreadCount }}."

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: "NiFi Process Group restart recovery gate complete for {{
      inputs.process_group_id }}: severity {{ outputs.evaluate.values.severity
      }}, confirm_start={{ inputs.confirm_start }}."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'NiFi Process Group restart recovery gate failed (execution ' ~ execution.id ~ '). Check the logs.'} | toJson }}

triggers:
  - id: manual_restart_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled and always in safe mode (confirm_start false) by
      default. Enable only if you want a periodic probe-only health check;
      trigger a manual execution with confirm_start true for the actual restart.
    cron: "0 * * * *"
    disabled: true
    inputs:
      nifi_url: "https://localhost:8443"
      nifi_username: "admin"
      ssl_verify: false
      process_group_id: "root"
      confirm_start: false
      min_active_threads_after_start: 1

Apache NiFi's Start API call tells you the request was accepted, not that the Process Group is actually processing anything again; a group can report 200 OK on start and still sit at zero active threads if the underlying stuck component was never really fixed. This blueprint treats the restart as unconfirmed until a fresh status read proves it, instead of trusting the API response alone.

This blueprint was created by Mohit Jeswani.

How it works

probe_before (io.kestra.plugin.nifi.GetProcessGroupStatus) reads the current queuedCount and activeThreadCount before anything runs.
route_confirm (io.kestra.plugin.core.flow.If) only calls io.kestra.plugin.nifi.StartProcessGroup when confirm_start is explicitly true; otherwise it logs the current state and changes nothing in NiFi.
probe_after (io.kestra.plugin.nifi.GetProcessGroupStatus) re-reads status right after, regardless of which branch ran, so evaluate always has a current reading.
evaluate (io.kestra.plugin.core.output.OutputValues) classifies DRY_RUN when nothing was started, CRITICAL when the post-start probe itself failed, RECOVERY_CONFIRMED when activeThreadCount reached min_active_threads_after_start, otherwise RECOVERY_NOT_CONFIRMED.
route_result (io.kestra.plugin.core.flow.If):
- compliant: logs DRY_RUN or RECOVERY_CONFIRMED with the current numbers.
- alert: posts CRITICAL or RECOVERY_NOT_CONFIRMED to Slack.
summary logs the final severity, closing the governance cycle.
errors alerts Slack if the flow itself fails, for example an expired NiFi token.

What you get

A restart that is verified, not assumed: the Start API call succeeding and the group actually processing again are checked as two different things.
A safe default: confirm_start must be explicitly set to true for anything in NiFi to change, every other run is read-only.
A fully local test path against the official Apache NiFi Docker image, no NiFi cluster needed.

Who it's for

Data engineering and platform teams running Apache NiFi who restart a Process Group after fixing a stuck component and want proof it actually recovered, not just that the API call did not error.
Teams evaluating the NiFi plugin who want a worked example of StartProcessGroup paired with a verification probe, the companion half of this repo's existing NiFi backlog and stall sentinel, which stops a group but deliberately leaves starting it back up as a separate, explicit flow run, this one.

Why orchestrate this with Kestra

NiFi's Start API reports whether the request was accepted, not whether the fix actually worked; confirming that takes a second read and a human (or a flow) comparing the two. Kestra keeps confirm_start as an explicit flow input instead of a habit of trusting the API response, turns an unconfirmed restart into a Slack page naming the real numbers, and keeps a full execution history of every attempt.

Prerequisites

For local testing, the official Apache NiFi Docker image with single-user authentication:
```
docker run --name nifi -d -p 8443:8443 \\
  -e SINGLE_USER_CREDENTIALS_USERNAME=admin \\
  -e SINGLE_USER_CREDENTIALS_PASSWORD="<a-password-at-least-12-characters>" \\
  apache/nifi:latest
```
NiFi's single-user password must be at least 12 characters or the image generates a random one instead and logs it, check the container logs if login fails. The UI and REST API are both served over HTTPS on port 8443 with a self-signed certificate by default, which is why ssl_verify defaults to false here.
A Process Group already stopped (for example by this repo's backlog and stall sentinel, or manually in the NiFi canvas) to exercise confirm_start: true against.
For real NiFi, a user with permission to view and modify the target Process Group's state.
A Slack incoming webhook URL.

Secrets

NIFI_PASSWORD: password for nifi_username. In Kestra OSS, provide it as the base64-encoded environment variable SECRET_NIFI_PASSWORD.
SLACK_WEBHOOK_URL: used by alert and the errors block. In Kestra OSS, provide it as the base64-encoded environment variable SECRET_SLACK_WEBHOOK_URL.

Inputs

Input	Default	Purpose
`process_group_id`	`root`	Process Group to restart and verify
`confirm_start`	`false`	Must be `true` for `start_process_group` to actually run
`min_active_threads_after_start`	`1`	Minimum `activeThreadCount` required for `RECOVERY_CONFIRMED`

Outputs

{{ outputs.probe_before.queuedCount }} / .activeThreadCount: state before any action.
{{ outputs.probe_after.queuedCount }} / .activeThreadCount: state immediately after.
{{ outputs.evaluate.values.severity }}: DRY_RUN, RECOVERY_CONFIRMED, RECOVERY_NOT_CONFIRMED, or CRITICAL.

Pitfalls

probe_after reads status immediately after start_process_group returns; NiFi may take a few seconds to actually spin threads up on a group with many components, so a single RECOVERY_NOT_CONFIRMED immediately after a real start can be spin-up latency rather than a genuine failure. Rerun this flow a minute later before concluding the fix did not work.
StartProcessGroup starts every component in the targeted Process Group; if the stuck component itself still errors immediately after starting, activeThreadCount can still read 0 even though the API call succeeded, which is exactly the case this gate is built to catch.
This flow never stops a group; it only starts one and verifies the result. Use this repo's NiFi backlog and stall sentinel to detect and pause a stuck group in the first place.

Quick start

Start the NiFi container above and add the NIFI_PASSWORD and SLACK_WEBHOOK_URL secrets.
Stop a Process Group (via this repo's backlog sentinel, or manually in the NiFi canvas).
Run this flow with confirm_start: false and confirm log_dry_run reports the stopped state without changing anything.
Rerun with confirm_start: true and confirm compliant or alert reports RECOVERY_CONFIRMED or RECOVERY_NOT_CONFIRMED matching the real outcome.

How to extend

Loop this flow over a list of Process Group IDs with io.kestra.plugin.core.flow.Loop to restart and verify several groups fixed in the same maintenance window.
Chain this flow as a follow-up execution triggered from the backlog and stall sentinel's own Slack alert, once an operator confirms the stuck component is fixed.
Add a second, later-scheduled run with confirm_start: false to catch a group that showed RECOVERY_CONFIRMED immediately but stalled again shortly after.

NiFi Process Group Restart Recovery Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Outputs

Pitfalls

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

NiFi Process Group Restart Recovery Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Outputs

Pitfalls

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

NetBox Drift Reconciliation with GitHub Audit Trail

Audit NetworkPolicy Coverage Across Every Namespace

New Relic APM Apdex Score Degradation Sentinel

Run Ansible Against NetBox Devices by Role and Site

Netskope Security Alert Volume Sentinel

Netlify Deploy Failure Sentinel

New to Kestra?