Kafka Consumer Lag Monitoring with Kestra

id: kafka-consumer-lag-alert
namespace: company.team
description: |
  Measure a Kafka consumer group's backlog with the broker's own tooling,
  and alert Slack the moment lag crosses your threshold, the group stops
  committing, or the group was never registered at all.

triggers:
  - id: lag_check_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check lag every five minutes. Shipped disabled; enable it after
      your first successful manual run so the inputs and threshold are proven
      before alerts start firing.
    cron: "*/5 * * * *"
    disabled: true

inputs:
  - id: bootstrap_servers
    type: STRING
    defaults: localhost:9092
    description: Bootstrap address of the Kafka cluster, reachable from the task
      container (host:port, comma separated for multiple brokers).
  - id: consumer_group
    type: STRING
    defaults: orders-consumer
    description: The consumer group whose backlog should be measured.
  - id: max_lag
    type: INT
    defaults: 10000
    description: Alert when the sum of lag across the group's partitions reaches or
      exceeds this many messages.
  - id: kafka_image
    type: STRING
    defaults: apache/kafka:3.9.0
    description: Container image that carries kafka-consumer-groups.sh. Any official
      Apache Kafka image works.

tasks:
  - id: report_lag
    type: io.kestra.plugin.scripts.shell.Commands
    description: Ask the broker for the group's committed offsets versus the end of
      each partition, and publish the totals — total lag, worst partition,
      partition count, active members, and whether the group exists at all —
      through Kestra's stdout outputs protocol. The missing-group case is
      detected from the tool's message because kafka-consumer-groups.sh exits 0
      even when the group is unknown.
    containerImage: "{{ inputs.kafka_image }}"
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      networkMode: host
    commands:
      - |
        set -u
        OUT=$(/opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server {{ inputs.bootstrap_servers }} --describe --group {{ inputs.consumer_group }} 2>&1)
        printf '%s\n' "$OUT"
        if printf '%s' "$OUT" | grep -q "does not exist"; then
          printf '::{"outputs":{"group_exists":false,"total_lag":0,"max_partition_lag":0,"partitions":0,"active_members":0}}::\n'
          exit 0
        fi
        printf '%s\n' "$OUT" | awk -v grp="{{ inputs.consumer_group }}" '
          $1 == "GROUP" { inside = 1; next }
          inside && $1 == grp {
            lag = $6 + 0
            total += lag
            if (lag > max) max = lag
            n += 1
            if ($7 != "-") members += 1
          }
          END {
            printf "::{\"outputs\":{\"group_exists\":true,\"total_lag\":%.0f,\"max_partition_lag\":%.0f,\"partitions\":%.0f,\"active_members\":%.0f}}::\n", total, max, n, members
          }'

  - id: check_group_missing
    type: io.kestra.plugin.core.flow.If
    description: A group that the broker has never heard of means the consumer has
      never committed — the check is pointed at the wrong name or the service
      never started, and silence would look like a healthy empty backlog.
    condition: "{{ outputs.report_lag.vars.group_exists == false }}"
    then:
      - id: alert_unknown_group
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Name the unknown group and the cluster so the stream owner can fix
          the name or start the consumer instead of trusting a quiet dashboard.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "Kafka lag check: consumer group {{ inputs.consumer_group }} does not exist on {{ inputs.bootstrap_servers }} — it has never committed an offset. Check the group name or start the consumer. Execution {{ execution.id }}."
          }

  - id: check_lag_threshold
    type: io.kestra.plugin.core.flow.If
    description: Sum the lag the report produced and compare it against the alert
      threshold — the single number that decides whether the backlog is a
      problem yet.
    condition: "{{ outputs.report_lag.vars.group_exists == true and
      outputs.report_lag.vars.total_lag >= inputs.max_lag }}"
    then:
      - id: alert_lag_breach
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Post total lag, the worst partition, how many partitions carry it,
          and how many consumers are attached, so the responder can tell a slow
          consumer from a dead one.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "Kafka lag ALERT: group {{ inputs.consumer_group }} on {{ inputs.bootstrap_servers }} is at {{ outputs.report_lag.vars.total_lag }} messages behind (threshold {{ inputs.max_lag }}). Worst partition: {{ outputs.report_lag.vars.max_partition_lag }}. Spread across {{ outputs.report_lag.vars.partitions }} partition(s) with {{ outputs.report_lag.vars.active_members }} active member(s). Execution {{ execution.id }}."
          }

  - id: check_stalled_consumer
    type: io.kestra.plugin.core.flow.If
    description: Lag below the threshold is not always fine — if there is any
      backlog at all and not a single consumer is attached to the group, nothing
      will ever drain it, so escalate on that shape too.
    condition: "{{ outputs.report_lag.vars.group_exists == true and
      outputs.report_lag.vars.total_lag > 0 and
      outputs.report_lag.vars.active_members == 0 }}"
    then:
      - id: alert_stalled_consumer
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Flag the orphaned backlog — a group with messages waiting and zero
          members means the consumer died or scaled to zero, and the lag will
          only grow.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "Kafka lag ALERT: group {{ inputs.consumer_group }} on {{ inputs.bootstrap_servers }} has {{ outputs.report_lag.vars.total_lag }} message(s) waiting but ZERO active consumers — the backlog will not drain on its own. Restart or scale the consumer. Execution {{ execution.id }}."
          }
    else:
      - id: log_healthy
        type: io.kestra.plugin.core.log.Log
        description: Record a clean reading so the execution history doubles as a lag
          timeline — the trend is as useful as the alerts.
        message: "Kafka group {{ inputs.consumer_group }} healthy: {{
          outputs.report_lag.vars.total_lag }} lag across {{
          outputs.report_lag.vars.partitions }} partition(s), worst {{
          outputs.report_lag.vars.max_partition_lag }}, {{
          outputs.report_lag.vars.active_members }} active member(s) (threshold
          {{ inputs.max_lag }})."

errors:
  - id: alert_check_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert when the measurement itself fails — an unreachable broker or
      a broken tool produces no numbers, and a check that dies quietly is
      indistinguishable from a healthy cluster.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "Kafka lag check FAILED in flow {{ flow.id }} (execution {{ execution.id }}) for group {{ inputs.consumer_group }} on {{ inputs.bootstrap_servers }}. Lag was NOT measured — verify the broker address, the Docker network, and that the image in kafka_image is pullable."
      }

outputs:
  - id: lag_summary
    type: JSON
    description: 'The measured backlog — e.g. {"group_exists": true, "total_lag": 8,
      "max_partition_lag": 8, "partitions": 1, "active_members": 0} — ready for
      dashboards or downstream flows.'
    value: "{{ outputs.report_lag.vars | toJson }}"

Lag is the heartbeat of a streaming pipeline: healthy, it stays flat; dying, it climbs until orders stop processing and someone asks why the dashboard is frozen. This blueprint turns the broker's own answer — kafka-consumer-groups.sh, the tool every Kafka operator already trusts — into a scheduled, branching decision. Each run computes total lag, the worst partition, and how many consumers are actually attached, then picks between three outcomes: over threshold, orphaned backlog, or quiet and healthy.

How it works

report_lag (io.kestra.plugin.scripts.shell.Commands on the io.kestra.plugin.scripts.runner.docker.Docker runner with networkMode: host) runs kafka-consumer-groups.sh --describe inside the official Apache Kafka image, parses the LAG column with awk, and emits group_exists, total_lag, max_partition_lag, partitions, and active_members through the ::{"outputs": ...}:: protocol. A missing group is detected from the tool's own message, because the command exits 0 even when the group is unknown.
check_group_missing (io.kestra.plugin.core.flow.If) catches the group the broker has never heard of — wrong name or consumer never started — and alerts instead of reporting a deceptively clean zero.
check_lag_threshold (io.kestra.plugin.core.flow.If) compares total lag against the max_lag input and posts the backlog, worst partition, and attached-member count to Slack when breached.
check_stalled_consumer (io.kestra.plugin.core.flow.If) escalates a shape thresholds miss: any messages waiting with zero active members means nothing will ever drain them. Under a clean reading, log_healthy writes the numbers to the execution history.
The errors block alerts when the measurement itself fails, and a five-minute Schedule (shipped disabled) drives the cadence once enabled.

What you get

A single trustworthy number — total lag — computed the same way every run, straight from the broker.
Three distinct alerts instead of one ambiguous page: over threshold, unknown group, orphaned backlog.
A lag_summary JSON output (total_lag, max_partition_lag, partitions, active_members) for dashboards or downstream flows.
An execution history that doubles as a lag timeline, with healthy runs logged too.

Who it's for

SREs and platform teams running order pipelines, CDC streams, or event ingestion where lag turns into customer-visible delay.
Streaming teams who want an alert before the backlog is obvious on a dashboard.
Anyone who has typed kafka-consumer-groups.sh --describe by hand at 2am and wished it paged them instead.

Why orchestrate this with Kestra

A cron job could echo the lag, but it could not branch three ways on what it found, alert only on the dangerous shapes, self-report its own failures, and keep every reading in a queryable execution history. Kestra supplies the schedule, the conditional logic, the failure channel, and the audit trail — and because the check is a flow, the same alert can later fan out to PagerDuty or open a Jira ticket by adding one task.

Prerequisites

A reachable Kafka cluster — the broker's bootstrap address as seen from the Kestra task runner (with networkMode: host, localhost:9092 covers a broker on the same host; containers on a shared Docker network use their service name).
Docker available to the Kestra Worker for the task runner.
A Slack incoming webhook for alerts.
The consumer group you intend to watch must have committed at least one offset (or you will — correctly — get the unknown-group alert).

Secrets

SLACK_WEBHOOK_URL: Slack incoming webhook URL for breach, missing-group, stalled-consumer, and failure alerts.

Quick start

Add the Slack webhook secret to your Kestra namespace.
Set bootstrap_servers and consumer_group to your cluster and group.
Run the flow once and read lag_summary (and the logs) to confirm the numbers match kafka-consumer-groups.sh --describe by hand.
Set disabled: false on the lag_check_schedule trigger and tune max_lag to your traffic.

How to extend

Add a second threshold input for a "warning" band so Slack gets a heads-up before the page.
Loop over several consumer groups by wrapping report_lag and the checks in a ForEach over a groups input.
Feed lag_summary into a Kestra dashboard or a nightly trend report by adding an output-consuming subflow.
Route the breach branch to PagerDuty or an incident tool by adding a second notification task beside Slack.

Watch a Kafka Consumer Group's Backlog and Alert Before It Becomes an Outage

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Watch a Kafka Consumer Group's Backlog and Alert Before It Becomes an Outage

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Renew TLS certificates through ACME, verify them, deploy them and roll back on failure

Autonomous Database Incident Investigation Agent with DuckDB and AI

AI Security Incident Threat Advisor and MITRE ATT&CK Containment Generator

Autonomous Incident Triage with Multi-Provider LLM Fallback and Slack Escalation

High-Performance Computing, Run Parallel Python Workloads on AWS Batch with ECS Fargate

More Data Blueprints

Kafka Topic Replication Sentinel

Real-Time Kafka Record Processing with Per-Record Executions

Kafka Consumer Lag Sentinel

Validate and Reshape a Webhook Order Payload with JSONata

Kafka Producer Compression Ratio Sentinel

Kafka DLQ Quarantine, Classification, and Safe Replay Gate

New to Kestra?