NiFi Process Group Backlog and Stall Sentinel with Kestra

id: nifi-process-group-backlog-stall-sentinel
namespace: company.team

description: |
  Poll an Apache NiFi Process Group's aggregate status, classify it HEALTHY, BACKLOG_WARNING,
  BACKLOG_CRITICAL, STALLED or PROBE_FAILED from the real queued FlowFile count and active
  thread count, alert Slack, and optionally pause the process group to stop a runaway backlog
  from growing further while a human investigates the stuck component.

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 audit, or the literal value root for the
      root Process Group. Remediation targets this exact group, so use a
      specific child group ID in production rather than root to avoid pausing
      the entire canvas.
    defaults: "root"

  - id: queued_count_warning_threshold
    type: INT
    displayName: Queued FlowFile warning threshold
    description: Aggregate queuedCount at or above which the group is classified
      BACKLOG_WARNING instead of HEALTHY.
    defaults: 100

  - id: queued_count_critical_threshold
    type: INT
    displayName: Queued FlowFile critical threshold
    description: Aggregate queuedCount at or above which the group is classified
      BACKLOG_CRITICAL.
    defaults: 1000

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    description: When true, together with dry_run false, stops this one Process
      Group if it is found STALLED or BACKLOG_CRITICAL, then verifies the stop
      took.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_remediate true, for the stop to
      actually run.
    defaults: true

tasks:
  - id: probe_process_group_status
    type: io.kestra.plugin.nifi.GetProcessGroupStatus
    description: Reads the aggregate queuedCount, queuedBytes, and activeThreadCount
      for process_group_id.
    allowFailure: true
    url: "{{ inputs.nifi_url }}"
    username: "{{ inputs.nifi_username }}"
    password: "{{ secret('NIFI_PASSWORD') }}"
    sslVerify: "{{ inputs.ssl_verify }}"
    processGroupId: "{{ inputs.process_group_id }}"

  - id: classify_process_group_status
    type: io.kestra.plugin.core.debug.Return
    description: PROBE_FAILED when the probe did not return at all, STALLED when
      FlowFiles are queued but nothing is actively processing them,
      BACKLOG_CRITICAL or BACKLOG_WARNING from the queued count against the two
      thresholds, HEALTHY otherwise.
    format: "{{ outputs.probe_process_group_status is not defined ? 'PROBE_FAILED' :
      ( ((outputs.probe_process_group_status.activeThreadCount is defined ?
      outputs.probe_process_group_status.activeThreadCount : 0) == 0 and
      (outputs.probe_process_group_status.queuedCount is defined ?
      outputs.probe_process_group_status.queuedCount : 0) >=
      inputs.queued_count_warning_threshold) ? 'STALLED' : (
      (outputs.probe_process_group_status.queuedCount is defined ?
      outputs.probe_process_group_status.queuedCount : 0) >=
      inputs.queued_count_critical_threshold ? 'BACKLOG_CRITICAL' : (
      (outputs.probe_process_group_status.queuedCount is defined ?
      outputs.probe_process_group_status.queuedCount : 0) >=
      inputs.queued_count_warning_threshold ? 'BACKLOG_WARNING' : 'HEALTHY' ) )
      ) }}"

  - id: route_by_process_group_status
    type: io.kestra.plugin.core.flow.Switch
    description: Branch A logs a healthy group. Branch B alerts Slack with the real
      queuedCount and activeThreadCount that triggered it, never an invented
      number.
    value: "{{ outputs.classify_process_group_status.value }}"
    cases:
      "HEALTHY":
        - id: log_healthy
          type: io.kestra.plugin.core.log.Log
          message: "NiFi Process Group {{ inputs.process_group_id }} healthy.
            queuedCount={{ outputs.probe_process_group_status.queuedCount }}
            activeThreadCount={{
            outputs.probe_process_group_status.activeThreadCount }}"
      "BACKLOG_WARNING":
        - id: alert_backlog_warning
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "NiFi Process Group " ~ inputs.process_group_id ~ " BACKLOG_WARNING. queuedCount=" ~ outputs.probe_process_group_status.queuedCount ~ ", at or above the " ~ inputs.queued_count_warning_threshold ~ " warning threshold."} | toJson }}
      "BACKLOG_CRITICAL":
        - id: alert_backlog_critical
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "NiFi Process Group " ~ inputs.process_group_id ~ " BACKLOG_CRITICAL. queuedCount=" ~ outputs.probe_process_group_status.queuedCount ~ ", at or above the " ~ inputs.queued_count_critical_threshold ~ " critical threshold. activeThreadCount=" ~ outputs.probe_process_group_status.activeThreadCount ~ "."} | toJson }}
      "STALLED":
        - id: alert_stalled
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "NiFi Process Group " ~ inputs.process_group_id ~ " STALLED. queuedCount=" ~ outputs.probe_process_group_status.queuedCount ~ " with activeThreadCount=0, FlowFiles are queued but nothing is processing them."} | toJson }}
      "PROBE_FAILED":
        - id: alert_probe_failed
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "NiFi Process Group status probe for " ~ inputs.process_group_id ~ " did not return a result. Check nifi_url, process_group_id, and NiFi credentials."} | toJson }}
    defaults:
      - id: log_unexpected_status
        type: io.kestra.plugin.core.log.Log
        message: "Unexpected classify_process_group_status value: {{
          outputs.classify_process_group_status.value }}"

  - id: remediate_gate
    type: io.kestra.plugin.core.flow.If
    description: Runs only when the group was STALLED or BACKLOG_CRITICAL,
      auto_remediate is true, and dry_run is false.
    condition: "{{ (outputs.classify_process_group_status.value == 'STALLED' or
      outputs.classify_process_group_status.value == 'BACKLOG_CRITICAL') and
      inputs.auto_remediate and not inputs.dry_run }}"
    then:
      - id: stop_process_group
        type: io.kestra.plugin.nifi.StopProcessGroup
        description: Schedules every component in this one Process Group to STOPPED,
          pausing further ingestion. This does not clear the existing queue and
          does not try to restart the specific stuck component, the plugin
          exposes no such task.
        allowFailure: true
        url: "{{ inputs.nifi_url }}"
        username: "{{ inputs.nifi_username }}"
        password: "{{ secret('NIFI_PASSWORD') }}"
        sslVerify: "{{ inputs.ssl_verify }}"
        processGroupId: "{{ inputs.process_group_id }}"

      - id: verify_process_group_stopped
        type: io.kestra.plugin.nifi.GetProcessGroupStatus
        description: Re-reads the group's status to confirm activeThreadCount dropped to
          0 after the stop.
        allowFailure: true
        url: "{{ inputs.nifi_url }}"
        username: "{{ inputs.nifi_username }}"
        password: "{{ secret('NIFI_PASSWORD') }}"
        sslVerify: "{{ inputs.ssl_verify }}"
        processGroupId: "{{ inputs.process_group_id }}"

      - id: report_remediation_result
        type: io.kestra.plugin.core.flow.If
        description: Fails the flow when the verify probe did not return at all or still
          reports active threads, so a stop that silently failed is never
          reported as fixed.
        condition: "{{ outputs.verify_process_group_stopped is not defined or
          (outputs.verify_process_group_stopped.activeThreadCount is defined ?
          outputs.verify_process_group_stopped.activeThreadCount : 1) != 0 }}"
        then:
          - id: remediation_not_confirmed
            type: io.kestra.plugin.core.execution.Fail
            errorMessage: "StopProcessGroup ran for Process Group {{ inputs.process_group_id
              }}, but it still reports active threads."
        else:
          - id: remediation_confirmed
            type: io.kestra.plugin.core.log.Log
            message: "Process Group {{ inputs.process_group_id }} stopped and verified
              idle."

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    description: Final audit log confirming the gate ran, regardless of the branch
      taken above.
    message: "NiFi Process Group backlog and stall sentinel completed.
      processGroupId={{ inputs.process_group_id }} status={{
      outputs.classify_process_group_status.value }} auto_remediate={{
      inputs.auto_remediate }} dry_run={{ inputs.dry_run }}"

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {{ {"text": "NiFi Process Group backlog and stall sentinel flow failed (execution " ~ execution.id ~ "). Check the execution logs."} | toJson }}

triggers:
  - id: every_10_minutes
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled, and always in safe mode (dry_run true) by default.
    cron: "*/10 * * * *"
    disabled: true
    inputs:
      nifi_url: "https://localhost:8443"
      nifi_username: "admin"
      ssl_verify: false
      process_group_id: "root"
      queued_count_warning_threshold: 100
      queued_count_critical_threshold: 1000
      auto_remediate: false
      dry_run: true

A NiFi Process Group can fill its internal queues for two very different reasons: the pipeline is simply busier than usual, or a downstream processor stopped, errored, or got stuck, and nothing is draining the queue at all. NiFi's own UI shows both as a growing queue count, the difference only becomes obvious once someone opens the canvas and checks whether any threads are actually active. This blueprint polls that distinction directly, alerts Slack with the real numbers that triggered it, and can optionally pause the group so an already-stuck backlog stops growing while a human finds the stuck component.

How it works

probe_process_group_status (io.kestra.plugin.nifi.GetProcessGroupStatus) reads the aggregate queuedCount, queuedBytes, and activeThreadCount for process_group_id, with allowFailure: true so a bad URL or credential does not stop the flow before it can be classified as PROBE_FAILED below.
classify_process_group_status (io.kestra.plugin.core.debug.Return) returns PROBE_FAILED when the probe did not return at all, STALLED when activeThreadCount is 0 while queuedCount is already at or above queued_count_warning_threshold (FlowFiles queued, nothing processing them), BACKLOG_CRITICAL or BACKLOG_WARNING from queuedCount against the two thresholds, HEALTHY otherwise.
route_by_process_group_status (io.kestra.plugin.core.flow.Switch) branches on that status: HEALTHY logs it, the other four each alert Slack with the real queuedCount and activeThreadCount that triggered it, never an invented number. An unexpected value falls through to a defaults branch that logs it rather than being silently dropped.
remediate_gate (io.kestra.plugin.core.flow.If) runs only when the status was STALLED or BACKLOG_CRITICAL, auto_remediate is true, and dry_run is false. It calls io.kestra.plugin.nifi.StopProcessGroup on this one process_group_id, then calls GetProcessGroupStatus again to confirm activeThreadCount actually dropped to 0, failing the flow with io.kestra.plugin.core.execution.Fail if the stop is not confirmed rather than silently reporting success.
log_audit_summary always runs last and logs the final classification.
The errors block posts to Slack if the flow itself fails outside the gate logic above.

What you get

A real distinction between "busy" and "stuck", instead of a single queue-depth number that looks the same for both.
A Slack alert carrying the exact queuedCount and activeThreadCount that triggered it, never an invented number.
An optional, verified pause: stopping a Process Group is confirmed by re-reading its status afterward, a stop that silently failed is never reported as fixed.
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 need to know the moment a dataflow stalls, not just when its queue eventually overflows.
Teams evaluating the NiFi plugin who want a working example of GetProcessGroupStatus, StopProcessGroup, and the plugin's JWT-based authentication.

Why orchestrate this with Kestra

NiFi's own UI and REST API expose queuedCount and activeThreadCount, but nothing in NiFi itself alerts on the combination, that logic would otherwise live in a bespoke monitoring script polling the REST API on a timer. Kestra replaces that script with a declarative flow: the schedule, the stall-versus-backlog classification, the Slack alert, and the optional verified pause are all one auditable execution history.

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.
To see BACKLOG_WARNING/BACKLOG_CRITICAL/STALLED rather than HEALTHY, build a simple flow in the NiFi canvas (for example GenerateFlowFile feeding a processor you then stop) so FlowFiles queue up without draining.
For real NiFi, a user with permission to view flow status and, if auto_remediate is used, permission to modify the target Process Group's state.
A Slack incoming webhook for alerts.

Secrets

NIFI_PASSWORD: password for nifi_username. The plugin's own username property is not a secret field, only password is.
SLACK_WEBHOOK: Slack incoming webhook URL used by the alert tasks and the errors block.

In Kestra OSS, set this as a base64-encoded environment variable, for example SECRET_NIFI_PASSWORD.

Inputs

Input	Type	Default	Allowed values
`nifi_url`	STRING	`https://localhost:8443`	any reachable NiFi base URL
`nifi_username`	STRING	`admin`	any NiFi username
`ssl_verify`	BOOL	`false`	`true`, `false`
`process_group_id`	STRING	`root`	a Process Group UUID, or `root`
`queued_count_warning_threshold`	INT	`100`	any non-negative integer
`queued_count_critical_threshold`	INT	`1000`	any non-negative integer, greater than the warning threshold
`auto_remediate`	BOOL	`false`	`true`, `false`
`dry_run`	BOOL	`true`	`true`, `false`

Quick start

Start the NiFi container above and add the NIFI_PASSWORD and SLACK_WEBHOOK secrets.
Run this flow manually with the defaults against the root group, it should report HEALTHY on a freshly started, empty canvas.
Build a small flow that queues FlowFiles without draining them, lower queued_count_warning_threshold to match, and run again, it should report BACKLOG_WARNING or STALLED depending on whether you also stopped the downstream processor.
Set auto_remediate: true and dry_run: false and run it again while STALLED or BACKLOG_CRITICAL, it should stop the group and log the verified confirmation.
Enable the every_10_minutes trigger, targeting a specific child Process Group ID rather than root if remediation should not be able to halt the entire canvas.

How to extend

Loop this flow over a list of Process Group IDs with io.kestra.plugin.core.flow.Loop to audit every major sub-flow in a large NiFi canvas from one flow.
Add queuedBytes to the classification alongside queuedCount, useful when a group's backlog is a small number of very large FlowFiles rather than many small ones.
Call io.kestra.plugin.nifi.StartProcessGroup after an operator has fixed the stuck component, as a second, explicitly separate flow run, rather than auto-restarting here.

Pitfalls

activeThreadCount is a single instantaneous sample; a healthy, bursty group can briefly read 0 between batches. STALLED only fires when queuedCount is also at or above queued_count_warning_threshold, reducing false positives from a momentary idle gap, but a single poll is still a snapshot, not a trend.
queuedCount, queuedBytes, and activeThreadCount come back null, treated here as 0, if NiFi's own response omits those keys, observed for a group with no components at all.
StopProcessGroup stops every component in the targeted Process Group, including the entire canvas if process_group_id is root, it does not and cannot restart just the one stuck processor, the plugin has no component-level task. Finding and fixing the actual stuck component still needs a human looking at NiFi's own bulletin board.
Stopping a group does not clear its existing queue, FlowFiles already queued stay queued until the group is manually started again after the root cause is fixed.

NiFi Process Group Backlog and Stall Sentinel

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

NiFi Process Group Backlog and Stall Sentinel

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Amazon SQS Message Age and Consumer Processing Lag Sentinel

AWS Transit Gateway Packet Drop Count Sentinel

AWS Site-to-Site VPN Tunnel State Down Sentinel

AWS WAF Blocked Requests Surge and L7 Attack Sentinel

Amazon SNS Subscription Delivery Failure and DLQ Sentinel

AWS VPC Flow Logs Rejected Traffic Anomaly Sentinel

New to Kestra?