Apache Druid Ingestion Task Failure & Duration Gate

id: druid-ingestion-task-failure-and-duration-gate
namespace: company.team

description: |
  Audit Apache Druid ingestion tasks from sys.tasks for unexpected task failures
  and stalled execution durations across streaming and batch ingestion jobs.
  Evaluates supervisor and task health, routes to live Slack notifications or
  audit-only logs, and maintains an end-to-end execution audit trail.

inputs:
  - id: druid_url
    type: STRING
    displayName: Druid Avatica JDBC URL
    description: JDBC Avatica connection URL for the Apache Druid router or broker
      service. Druid router listens on HTTP port 8888.
    defaults: "jdbc:avatica:remote:url=http://localhost:8888/druid/v2/sql/avatica/;\
      transparent_reconnection=true"

  - id: lookback_hours
    type: INT
    displayName: Ingestion Audit Lookback (Hours)
    description: Lookback window in hours to scan sys.tasks for completed or failed
      ingestion tasks.
    defaults: 2

  - id: max_allowed_failed_tasks
    type: INT
    displayName: Max Allowed Failed Tasks
    description: Threshold of failed ingestion tasks permitted within the lookback
      window before the gate trips into an alert state. Set to 1 to alert on any
      failure.
    defaults: 1

  - id: max_task_duration_ms
    type: INT
    displayName: Max Task Execution Duration (ms)
    description: Maximum allowable running duration in milliseconds for an active
      ingestion task. Tasks exceeding this duration are classified as stalled
      workers. Default is 2 hours (7200000 ms).
    defaults: 7200000

  - id: max_allowed_stalled_tasks
    type: INT
    displayName: Max Allowed Stalled Tasks
    description: Threshold of stalled running tasks exceeding max_task_duration_ms
      before triggering an operational alert.
    defaults: 1

  - id: alert_mode
    type: SELECT
    displayName: Sentinel Alert Mode
    description: Operating mode of the sentinel. ALERT dispatches a notification to
      Slack when thresholds breach; AUDIT_ONLY logs findings without paging
      on-call.
    values:
      - "ALERT"
      - "AUDIT_ONLY"
    defaults: "ALERT"

triggers:
  - id: periodic_sentinel_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: Runs every 30 minutes to evaluate ingestion task health and
      duration bounds. Shipped disabled by default for safe initial setup.
    cron: "*/30 * * * *"
    disabled: true
    inputs:
      druid_url: "jdbc:avatica:remote:url=http://localhost:8888/druid/v2/sql/avatica/\
        ;transparent_reconnection=true"
      lookback_hours: 2
      max_allowed_failed_tasks: 1
      max_task_duration_ms: 7200000
      max_allowed_stalled_tasks: 1
      alert_mode: "ALERT"

tasks:
  - id: probe_ingestion_tasks
    type: io.kestra.plugin.jdbc.druid.Query
    description: >
      Query Druid sys.tasks to aggregate total tasks, failed tasks, and stalled
      running tasks exceeding max_task_duration_ms within the configured
      lookback window.
    url: "{{ inputs.druid_url }}"
    username: "{{ secret('DRUID_USERNAME') }}"
    password: "{{ secret('DRUID_PASSWORD') }}"
    fetchType: FETCH_ONE
    sql: |
      SELECT
        COUNT(*) AS total_tasks,
        COALESCE(SUM(CASE WHEN status = 'FAILED' THEN 1 ELSE 0 END), 0) AS failed_task_count,
        COALESCE(SUM(CASE WHEN status = 'RUNNING' AND duration > {{ inputs.max_task_duration_ms }} THEN 1 ELSE 0 END), 0) AS stalled_task_count,
        COALESCE(MAX(CASE WHEN status = 'RUNNING' THEN duration ELSE 0 END), 0) AS max_running_duration_ms
      FROM sys.tasks
      WHERE created_time >= CURRENT_TIMESTAMP - INTERVAL '{{ inputs.lookback_hours }}' HOUR

  - id: probe_failed_task_offenders
    type: io.kestra.plugin.jdbc.druid.Query
    description: >
      Fetch the top 5 offending failed or stalled tasks with task ID,
      datasource, type, status, duration, and error message to provide
      actionable incident context.
    url: "{{ inputs.druid_url }}"
    username: "{{ secret('DRUID_USERNAME') }}"
    password: "{{ secret('DRUID_PASSWORD') }}"
    fetchType: FETCH
    sql: |
      SELECT
        task_id,
        datasource,
        type,
        status,
        duration,
        SUBSTRING(COALESCE(error_msg, 'No error message reported'), 1, 140) AS error_preview
      FROM sys.tasks
      WHERE (status = 'FAILED' OR (status = 'RUNNING' AND duration > {{ inputs.max_task_duration_ms }}))
        AND created_time >= CURRENT_TIMESTAMP - INTERVAL '{{ inputs.lookback_hours }}' HOUR
      ORDER BY duration DESC
      LIMIT 5

  - id: probe_worker_cluster_capacity
    type: io.kestra.plugin.jdbc.druid.Query
    description: >
      Query sys.servers to count active MiddleManager nodes and ensure worker
      slot capacity is available for task scheduling.
    url: "{{ inputs.druid_url }}"
    username: "{{ secret('DRUID_USERNAME') }}"
    password: "{{ secret('DRUID_PASSWORD') }}"
    fetchType: FETCH_ONE
    sql: |
      SELECT
        COUNT(*) AS total_servers,
        COALESCE(SUM(CASE WHEN server_type = 'middle_manager' THEN 1 ELSE 0 END), 0) AS middle_manager_count,
        COALESCE(SUM(CASE WHEN server_type = 'historical' THEN 1 ELSE 0 END), 0) AS historical_count
      FROM sys.servers

  - id: evaluate_ingestion_health
    type: io.kestra.plugin.core.flow.If
    description: >
      Evaluate aggregate probe metrics. Triggers the alert branch if failed task
      count meets or exceeds max_allowed_failed_tasks, or if stalled task count
      meets or exceeds max_allowed_stalled_tasks.
    condition: "{{ (outputs.probe_ingestion_tasks.row.failed_task_count >=
      inputs.max_allowed_failed_tasks) or
      (outputs.probe_ingestion_tasks.row.stalled_task_count >=
      inputs.max_allowed_stalled_tasks) }}"
    then:
      - id: evaluate_alert_mode
        type: io.kestra.plugin.core.flow.If
        description: >
          Check configured alert_mode. ALERT mode dispatches a Slack incoming
          webhook; AUDIT_ONLY mode logs structured findings without paging
          operators.
        condition: "{{ inputs.alert_mode == 'ALERT' }}"
        then:
          - id: send_slack_alert
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: >
              Post a structured alert notification to the designated Slack
              channel detailing failed tasks, stalled durations, and cluster
              capacity metrics.
            url: "{{ secret('SLACK_WEBHOOK') }}"
            payload: |
              {
                "text": "Druid ingestion gate triggered on {{ inputs.druid_url }}",
                "blocks": [
                  {
                    "type": "header",
                    "text": {
                      "type": "plain_text",
                      "text": "Apache Druid Ingestion Task Gate Triggered"
                    }
                  },
                  {
                    "type": "section",
                    "fields": [
                      {
                        "type": "mrkdwn",
                        "text": "*Failed Tasks ({{ inputs.lookback_hours }}h):*\n{{ outputs.probe_ingestion_tasks.row.failed_task_count }} (Limit: {{ inputs.max_allowed_failed_tasks }})"
                      },
                      {
                        "type": "mrkdwn",
                        "text": "*Stalled Running Tasks:*\n{{ outputs.probe_ingestion_tasks.row.stalled_task_count }} (Limit: {{ inputs.max_allowed_stalled_tasks }})"
                      },
                      {
                        "type": "mrkdwn",
                        "text": "*Max Running Duration:*\n{{ (outputs.probe_ingestion_tasks.row.max_running_duration_ms / 60000) | int }} min (Threshold: {{ (inputs.max_task_duration_ms / 60000) | int }} min)"
                      },
                      {
                        "type": "mrkdwn",
                        "text": "*Active MiddleManagers:*\n{{ outputs.probe_worker_cluster_capacity.row.middle_manager_count }}"
                      }
                    ]
                  },
                  {
                    "type": "section",
                    "text": {
                      "type": "mrkdwn",
                      "text": "*Execution:* {{ flow.namespace }}.{{ flow.id }} | ID: {{ execution.id }}"
                    }
                  }
                ]
              }
        else:
          - id: log_audit_only_breach
            type: io.kestra.plugin.core.log.Log
            description: >
              Emit structured audit finding when thresholds are breached in
              AUDIT_ONLY mode.
            message: >
              AUDIT_ONLY: Druid ingestion task gate breach detected. Failed
              tasks in last {{ inputs.lookback_hours }}h: {{
              outputs.probe_ingestion_tasks.row.failed_task_count }} (Limit: {{
              inputs.max_allowed_failed_tasks }}). Stalled running tasks: {{
              outputs.probe_ingestion_tasks.row.stalled_task_count }} (Limit: {{
              inputs.max_allowed_stalled_tasks }}). Max running duration: {{
              (outputs.probe_ingestion_tasks.row.max_running_duration_ms /
              60000) | int }} minutes.
    else:
      - id: log_healthy_status
        type: io.kestra.plugin.core.log.Log
        description: >
          Log nominal operational state when ingestion task metrics remain
          within thresholds.
        message: >
          HEALTHY: Apache Druid ingestion tasks operating normally. Total tasks
          in window: {{ outputs.probe_ingestion_tasks.row.total_tasks }}. Failed
          tasks: {{ outputs.probe_ingestion_tasks.row.failed_task_count }}.
          Stalled tasks: {{ outputs.probe_ingestion_tasks.row.stalled_task_count
          }}. MiddleManagers online: {{
          outputs.probe_worker_cluster_capacity.row.middle_manager_count }}.

  - id: log_sentinel_audit_complete
    type: io.kestra.plugin.core.log.Log
    description: >
      Emit a final audit summary confirming the sentinel cycle concluded
      successfully.
    message: >
      Sentinel cycle complete for flow {{ flow.id }} execution {{ execution.id
      }}. Evaluated failed count: {{
      outputs.probe_ingestion_tasks.row.failed_task_count }}. Evaluated stalled
      count: {{ outputs.probe_ingestion_tasks.row.stalled_task_count }}. Mode:
      {{ inputs.alert_mode }}.

errors:
  - id: on_failure_alert
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: >
      Flow-level error handler dispatched if Druid Avatica JDBC endpoint is
      unreachable or query execution fails.
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "text": "Apache Druid ingestion sentinel execution {{ execution.id }} FAILED in namespace {{ flow.namespace }}. Check Druid Avatica router connectivity and credentials."
      }

Apache Druid relies on distributed ingestion tasks running on MiddleManager workers or indexing peons to consume high-throughput streaming events from Kafka or Kinesis, as well as to execute scheduled compaction and batch loads. When data schema drift occurs, network partitions disrupt stream ingestion, or worker slots experience memory pressure, ingestion tasks can fail repeatedly or stall in a running state indefinitely, exhausting MiddleManager execution slots and stalling downstream analytical dashboards.

This blueprint provides an automated Day 2 operations sentinel for Apache Druid. It inspects Druid's native sys.tasks and sys.servers system catalog views over Avatica JDBC, evaluates aggregate failure counts and stalled task execution durations against operational SLA thresholds, and dispatches structured Slack notifications when ingestion health degrades.

How it works

periodic_sentinel_schedule triggers the sentinel workflow on a 30-minute cron schedule (disabled by default so connection endpoints and thresholds can be configured).
probe_ingestion_tasks executes an aggregating SQL query against sys.tasks via io.kestra.plugin.jdbc.druid.Query with fetchType: FETCH_ONE. It quantifies total tasks, failed tasks within the rolling lookback window, and active tasks exceeding max_task_duration_ms.
probe_failed_task_offenders queries sys.tasks with fetchType: FETCH to extract detailed context (task ID, datasource, task type, duration, and error preview) for up to 5 offending tasks.
probe_worker_cluster_capacity queries sys.servers to verify the count of healthy MiddleManager and Historical worker nodes currently registered in the cluster.
evaluate_ingestion_health uses io.kestra.plugin.core.flow.If to check if failed_task_count meets or exceeds max_allowed_failed_tasks, or if stalled_task_count meets or exceeds max_allowed_stalled_tasks.
When a breach occurs, evaluate_alert_mode checks alert_mode: ALERT posts a formatted Slack notification via io.kestra.plugin.slack.notifications.SlackIncomingWebhook, while AUDIT_ONLY logs the violation without paging on-call staff.
When performance is nominal, log_healthy_status records a healthy audit entry.
log_sentinel_audit_complete runs at the conclusion of every execution to maintain an unbroken audit trail of sentinel passes.
A flow-level errors handler catches connection errors or unreachable endpoints and alerts operators that the monitoring sentinel is offline.

What you get

Automated, continuous monitoring of Apache Druid streaming and batch ingestion tasks without deploying external Prometheus agents.
Dual-risk evaluation: detects both hard task termination (status = 'FAILED') and runaway resource starvation (status = 'RUNNING' past time ceiling).
Top offender context: captures offending task IDs, datasources, and root-cause error messages directly in execution outputs.
Dry-run capability via AUDIT_ONLY mode to benchmark baseline task turnover before enabling live paging channels.
Out-of-band error handling that catches network partitions or cluster downtime and prevents silent monitoring failures.

Who it's for

Data Platform Engineers and SREs operating Apache Druid clusters for real-time analytics and user-facing dashboards.
Streaming data pipeline owners maintaining Kafka and Kinesis ingestion supervisors.
Analytics Engineers requiring SLA guarantees on streaming ingestion freshness and compaction progress.

Why orchestrate this with Kestra

Druid's built-in coordinator and overlord web consoles display task states interactively but do not provide autonomous alerting or conditional incident triage workflows. Kestra delivers declarative, version-controlled workflow orchestration with built-in secret masking, conditional branching, native Slack integrations, and full execution history tracking. Managing the sentinel as a Kestra blueprint replaces brittle custom cron scripts with a resilient, observable automation pipeline.

Prerequisites

An Apache Druid cluster (version 0.23.0 or later, including 30.0.0+) reachable via HTTP/HTTPS over its Avatica JDBC SQL endpoint.
A Druid user account with permission to query system catalog tables (sys.tasks, sys.servers).
A Slack incoming webhook URL for dispatching alerts.

Secrets

DRUID_USERNAME: Database user for authenticating against Druid Avatica SQL endpoint.
DRUID_PASSWORD: Database password for authenticating against Druid Avatica SQL endpoint.
SLACK_WEBHOOK: Incoming webhook URL for the target Slack channel (required for live alerting).

Note for Kestra Open-Source (OSS): Secrets are configured as base64-encoded environment variables formatted as SECRET_<NAME> (e.g. SECRET_DRUID_USERNAME, SECRET_SLACK_WEBHOOK).

Inputs

Name	Type	Default	Allowed Values	Description
`druid_url`	STRING	`"jdbc:avatica:remote:url=http://localhost:8888/druid/v2/sql/avatica/;transparent_reconnection=true"`	Valid Avatica JDBC URL	Druid Avatica JDBC SQL connection URL
`lookback_hours`	INT	`2`	Positive integer	Lookback window in hours to scan `sys.tasks`
`max_allowed_failed_tasks`	INT	`1`	Positive integer	Maximum failed ingestion tasks allowed before alert
`max_task_duration_ms`	INT	`7200000`	Positive integer	Maximum allowable running duration in milliseconds
`max_allowed_stalled_tasks`	INT	`1`	Positive integer	Maximum stalled tasks allowed before alert
`alert_mode`	SELECT	`"ALERT"`	`"ALERT"`, `"AUDIT_ONLY"`	Operational alerting mode on threshold breach

Outputs

{{ outputs.probe_ingestion_tasks.row.failed_task_count }}: Aggregate count of failed tasks within the lookback window.
{{ outputs.probe_ingestion_tasks.row.stalled_task_count }}: Count of active tasks exceeding the duration ceiling.
{{ outputs.probe_ingestion_tasks.row.max_running_duration_ms }}: Maximum execution duration observed among active running tasks.
{{ outputs.probe_failed_task_offenders.rows }}: Structured list of top offending failed or stalled tasks with task ID, datasource, and error messages.
{{ outputs.probe_worker_cluster_capacity.row.middle_manager_count }}: Active MiddleManager nodes registered in the cluster.

Quick start

Start a local Apache Druid single-container instance on port 8888 (avoiding host port 8080):

docker run -d
--name druid-sentinel
-p 8888:8888
apache/druid:30.0.0
Connect the Druid container to the Kestra Docker network:

docker network connect YOUR_KESTRA_NETWORK druid-sentinel

Identify your network name using docker network ls.
Configure the DRUID_USERNAME, DRUID_PASSWORD, and SLACK_WEBHOOK secrets in your Kestra environment.
Import this blueprint into your Kestra instance.
Run the flow manually with alert_mode: AUDIT_ONLY to establish baseline task metrics and verify Avatica JDBC connectivity.
Enable periodic_sentinel_schedule to begin automated 30-minute operational sweeps.

How to extend

Add automated task remediation by issuing an HTTP POST request to Druid Overlord endpoint /druid/indexer/v1/task/{taskId}/shutdown to kill stalled runaway tasks.
Add partition and segment availability validation by joining with sys.segments to confirm published segments match queryable historical states.
Route critical streaming ingestion alerts to PagerDuty or Opsgenie while routing batch compaction delays to informational Slack channels.
Export execution audit metrics to BigQuery, Snowflake, or DuckDB for long-term ingestion SLA compliance trend analysis.

Pitfalls

Avatica URL format: Apache Druid requires the Avatica JDBC prefix with transparent reconnection enabled: jdbc:avatica:remote:url=http://<host>:8888/druid/v2/sql/avatica/;transparent_reconnection=true. Omitting the path suffix will cause connection timeouts.
Port conflict: Apache Druid Router listens on HTTP port 8888 by default. Never map container ports to host port 8080 to prevent collision with Kestra web UI.
Network isolation: When running Kestra and Druid in separate Docker containers, localhost:8888 will resolve to the Kestra container itself. Use the container hostname (e.g. http://druid-sentinel:8888) on a shared Docker bridge network.
SQL compatibility: Druid system tables use Calcite SQL semantics. String timestamps and intervals require standard ANSI formatting (e.g. CURRENT_TIMESTAMP - INTERVAL '2' HOUR).
Task retention: Tasks remain in sys.tasks for the duration configured by Druid task retention settings (druid.indexer.storage.recentlyFinishedThreshold). Ensure lookback_hours does not exceed this retention window.