Kafka Stale Consumer Group Cleanup Guard with Kestra

id: kafka-stale-consumer-group-cleanup-guard
namespace: company.team

description: |
  Find Kafka consumer groups that are Empty or Dead (no active members) and have stayed that
  way past a safety threshold, then delete only those behind an approval gate and re-list to
  confirm each one is gone. Dead applications leave empty groups in the coordinator and offset
  storage, where they clutter lag dashboards and never get cleaned. A group that is only briefly
  empty, a consumer restarting, is reported but never deleted, because an idle age is tracked per
  group in a KV cursor so the clock has to run out first.

inputs:
  - id: bootstrap_servers
    type: STRING
    displayName: Kafka bootstrap servers
    defaults: "localhost:9092"

  - id: min_idle_minutes
    type: INT
    displayName: Minimum minutes a group must stay empty before deletion
    description: A group empty for less than this is reported but never deleted, so
      a consumer that restarts within the window keeps its group.
    defaults: 1440

  - id: protected_groups
    type: ARRAY
    itemType: STRING
    displayName: Groups never deleted
    description: Group ids always kept even when empty, for example a group that a
      scheduled job recreates.
    defaults: []

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    defaults: false

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

variables:
  kv_key: "kafka_stale_groups_{{ inputs.bootstrap_servers | slugify }}"

tasks:
  - id: freeze
    type: io.kestra.plugin.core.output.OutputValues
    description: One timestamp for the whole run, so detect, cursor and age check agree.
    values:
      now_iso: >-
        {{ execution.startDate | date("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'",
        timeZone="UTC") }}
      cutoff_iso: >-
        {{ execution.startDate | dateAdd(-1 * inputs.min_idle_minutes,
        'MINUTES') | date("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'", timeZone="UTC") }}

  - id: list_all
    type: io.kestra.plugin.kafka.ConsumerGroupList
    description: Read-only. Every consumer group on the cluster with its state.
    properties:
      bootstrap.servers: "{{ inputs.bootstrap_servers }}"

  - id: candidates
    type: io.kestra.plugin.core.output.OutputValues
    values:
      empty_ids: >-
        {{ {'g': outputs.list_all.groups, 'prot': inputs.protected_groups} |
        jq('.prot as $p | [.g[] | select((.state == "Empty" or .state == "Dead")
        and (([.groupId] - $p) | length > 0)) | .groupId]') | first | toJson }}

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ fromJson(outputs.candidates.values.empty_ids) | length > 0 }}"
    then:
      - id: describe_candidates
        type: io.kestra.plugin.kafka.ConsumerGroupDescribe
        description: Read-only. Members and offsets of the empty candidates, to confirm
          zero active members before any delete.
        properties:
          bootstrap.servers: "{{ inputs.bootstrap_servers }}"
        groupIds: "{{ fromJson(outputs.candidates.values.empty_ids) }}"

      - id: confirmed
        type: io.kestra.plugin.core.output.OutputValues
        values:
          confirmed_ids: >-
            {{ outputs.describe_candidates.groups | jq('[.[] | select((.state ==
            "Empty" or .state == "Dead") and ((.members // []) | length == 0)) |
            .groupId]') | first | toJson }}

      - id: cursor_merge
        type: io.kestra.plugin.core.output.OutputValues
        description: Keep the first time each still-empty group was seen, add now for
          new ones, drop any group no longer empty.
        values:
          merged: >-
            {{ {'cur': (kv(render(vars.kv_key), errorOnMissing=false) ?? {}),
            'ids': fromJson(outputs.confirmed.values.confirmed_ids), 'now':
            outputs.freeze.values.now_iso} | jq('.cur as $c | .now as $n |
            reduce .ids[] as $id ({}; . + {($id): ($c[$id] // $n)})') | first |
            toJson }}

      - id: save_cursor
        type: io.kestra.plugin.core.kv.Set
        description: No ttl. The cursor is a first-seen record that must not expire out
          from under the age check.
        key: "{{ render(vars.kv_key) }}"
        kvType: JSON
        value: "{{ outputs.cursor_merge.values.merged }}"

      - id: targets
        type: io.kestra.plugin.core.output.OutputValues
        values:
          target_ids: >-
            {{ {'cur': fromJson(outputs.cursor_merge.values.merged), 'ids':
            fromJson(outputs.confirmed.values.confirmed_ids), 'cutoff':
            outputs.freeze.values.cutoff_iso} | jq('.cur as $c | .cutoff as $cut
            | [.ids[] | select(($c[.] // "9999") <= $cut)]') | first | toJson }}
          held_ids: >-
            {{ {'cur': fromJson(outputs.cursor_merge.values.merged), 'ids':
            fromJson(outputs.confirmed.values.confirmed_ids), 'cutoff':
            outputs.freeze.values.cutoff_iso} | jq('.cur as $c | .cutoff as $cut
            | [.ids[] | select(($c[.] // "9999") > $cut)]') | first | toJson }}

      - id: log_findings
        type: io.kestra.plugin.core.log.Log
        message: >-
          Stale consumer group scan on {{ inputs.bootstrap_servers }}: {{
          fromJson(outputs.confirmed.values.confirmed_ids) | length }} empty
          groups confirmed ({{ outputs.confirmed.values.confirmed_ids |
          jq('join(", ")') | first }}). {{
          fromJson(outputs.targets.values.target_ids) | length }} past the {{
          inputs.min_idle_minutes }} minute idle threshold ({{
          outputs.targets.values.target_ids | jq('join(", ")') | first }}). {{
          fromJson(outputs.targets.values.held_ids) | length }} held as too
          recently emptied ({{ outputs.targets.values.held_ids | jq('join(",
          ")') | first }}). auto_remediate={{ inputs.auto_remediate }}
          dry_run={{ inputs.dry_run }}.

      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: >-
          {{ {'text': 'Kafka ' ~ inputs.bootstrap_servers ~ ': ' ~
          (fromJson(outputs.confirmed.values.confirmed_ids) | length) ~ ' empty
          consumer groups, ' ~ (fromJson(outputs.targets.values.target_ids) |
          length) ~ ' stale past ' ~ inputs.min_idle_minutes ~ ' minutes (' ~
          (outputs.targets.values.target_ids | jq('join(", ")') | first) ~ ').
          auto_remediate=' ~ inputs.auto_remediate ~ ' dry_run=' ~
          inputs.dry_run ~ '.'} | toJson }}

      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run and
          (fromJson(outputs.targets.values.target_ids) | length > 0) }}"
        then:
          - id: approve
            type: io.kestra.plugin.core.flow.Pause
            description: Resume with approved=true to delete the groups listed by
              log_findings. Resume without it or kill the execution to cancel.
            onResume:
              - id: approved
                type: BOOL
                displayName: Approve deleting the stale empty consumer groups
                required: false

          - id: do_delete
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.approve.onResume.approved ?? false }}"
            then:
              - id: delete_groups
                type: io.kestra.plugin.kafka.ConsumerGroupDelete
                description: Deletes the targeted empty groups in one admin call. allowFailure
                  lets the re-list be the authority on what actually went.
                allowFailure: true
                properties:
                  bootstrap.servers: "{{ inputs.bootstrap_servers }}"
                groupIds: "{{ fromJson(outputs.targets.values.target_ids) }}"

              - id: reverify
                type: io.kestra.plugin.kafka.ConsumerGroupList
                description: Read-only re-list after the delete.
                properties:
                  bootstrap.servers: "{{ inputs.bootstrap_servers }}"

              - id: verify_result
                type: io.kestra.plugin.core.flow.If
                condition: >-
                  {{ {'now': outputs.reverify.groups, 'targets':
                  fromJson(outputs.targets.values.target_ids)} |
                  jq('[.now[].groupId] as $present | [.targets[] | select(([.] -
                  $present) | length == 0)] | length') | first > 0 }}
                then:
                  - id: delete_failed
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: >-
                      ConsumerGroupDelete ran, but a re-list still shows a
                      targeted group. Remaining: {{ {'now':
                      outputs.reverify.groups, 'targets':
                      fromJson(outputs.targets.values.target_ids)} |
                      jq('[.now[].groupId] as $present | [.targets[] |
                      select(([.] - $present) | length == 0)] | join(", ")') |
                      first }}. A member may have rejoined before the delete.
                else:
                  - id: deleted_ok
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      Deleted {{ fromJson(outputs.targets.values.target_ids) |
                      length }} stale empty consumer groups ({{
                      outputs.targets.values.target_ids | jq('join(", ")') |
                      first }}). A re-list confirms none remain.
            else:
              - id: not_approved
                type: io.kestra.plugin.core.log.Log
                message: "Pause resumed without approval or killed. No consumer group was
                  deleted."
    else:
      - id: clear_cursor
        type: io.kestra.plugin.core.kv.Set
        description: No empty groups, so reset the first-seen cursor.
        key: "{{ render(vars.kv_key) }}"
        kvType: JSON
        value: "{}"

      - id: all_active
        type: io.kestra.plugin.core.log.Log
        message: "No empty or dead consumer groups on {{ inputs.bootstrap_servers }}.
          Nothing to clean."

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: >-
      {{ {'text': 'Kafka stale consumer group cleanup guard failed on ' ~
      inputs.bootstrap_servers ~ ' (execution ' ~ execution.id ~ '). No group
      was deleted unless a log line says deleted. Check the logs.'} | toJson }}

triggers:
  - id: daily_scan
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled and always in safe mode.
    cron: "0 3 * * *"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true

Dead applications leave consumer groups behind. The last member leaves, the group goes to state Empty with its committed offsets still stored and nothing ever removes it. Over months a cluster collects hundreds of these. They clutter kafka-consumer-groups.sh --list, they show up on lag dashboards reading against offsets no one consumes and they make the real groups harder to find.

Deleting them by hand is risky. A group that looks empty right now might be a consumer that is restarting or rebalancing. Deleting its group drops the committed offsets, so when it comes back it reprocesses or skips messages. The safe rule is to delete a group only after it has stayed empty long enough that it is clearly abandoned.

This blueprint lists every consumer group, keeps the ones that are Empty or Dead with zero active members and tracks how long each has been empty in a KV cursor. A group deletes only when it has been empty past min_idle_minutes, only when auto_remediate is true and dry_run is false and only after someone approves the paused execution. After the delete it re-lists the groups and fails the run if any targeted group is still there.

This blueprint was created by zkasuran.

Safety

Read-only by default. With the shipped inputs it only lists, reports and alerts. It changes nothing.
A delete needs three things at once: auto_remediate true, dry_run false and a human resuming the Pause with approved=true. An unanswered resume reads as not approved.
Only groups confirmed Empty or Dead with zero members are ever targeted. Kafka itself refuses to delete a group with active members, which is a second net under the detector.
The age cursor has no ttl, so a group has to be seen empty across the whole window before it is eligible. A brief empty window from a restart never crosses the line.
protected_groups are never deleted even when empty.

How it works

freeze captures one timestamp for the run, so the detect, cursor and age steps all agree on now.
list_all (io.kestra.plugin.kafka.ConsumerGroupList, read-only) returns every group with its state.
candidates keeps the groups in state Empty or Dead that are not in protected_groups.
route (io.kestra.plugin.core.flow.If) branches on whether any candidate exists.
- When none exist, clear_cursor resets the first-seen record and all_active logs that there is nothing to clean. No alert.
- When some exist:
  - describe_candidates (io.kestra.plugin.kafka.ConsumerGroupDescribe, read-only) reads members and offsets, then confirmed keeps only the ones with zero active members.
  - cursor_merge and save_cursor keep the first time each still-empty group was seen in the KV key kafka_stale_groups_<cluster>, with no ttl.
  - targets splits the confirmed groups into those empty longer than min_idle_minutes and those held back as too recent.
  - log_findings logs the split and alert posts it to Slack.
  - remediate_gate runs only when auto_remediate is true, dry_run is false and at least one group is past the threshold.
    - approve (io.kestra.plugin.core.flow.Pause) waits for a resume carrying approved.
    - When approved, delete_groups (io.kestra.plugin.kafka.ConsumerGroupDelete) deletes the targeted groups, reverify re-lists and verify_result fails the run if any targeted group is still present.
    - When not approved, not_approved logs that nothing was deleted.
errors posts to Slack on any failure, for example an unreachable broker or a group that survived the delete.

What you get

A clean inventory of abandoned consumer groups, separated from the ones still in use.
An age gate that will not delete a group that was empty only for a moment.
Deletion that is off by default, gated behind auto_remediate, dry_run and a human approval.
Proof the cleanup landed: a re-list after the delete that fails the run if a target survived.

Who it's for

Platform teams running shared Kafka clusters where old services keep leaving groups behind.
On-call engineers who want the empty groups gone without typing --delete against a live cluster by hand.
Teams evaluating the Kafka plugin who want a worked example of ConsumerGroupList, ConsumerGroupDescribe and ConsumerGroupDelete with gates around the delete.

Why orchestrate this with Kestra

kafka-consumer-groups.sh --delete removes a group straight away. It does not remember how long the group has been empty, hold the delete for a second person or check afterwards that the group is really gone. Kestra adds the schedule that watches the groups, the KV cursor that measures the idle age across runs, the Pause that turns a delete into an approved change, the re-list that proves the result and a history of every cleanup.

Tested end to end

On Kestra 2.0.5 OSS against Apache Kafka 3.9.1 (KRaft, single node), with a local HTTP sink standing in for Slack. The topic orders had 3 partitions and 30 messages. Three groups (legacy-exporter, old-reporting, defunct-worker) read to the end then stopped, so each went to state Empty. One group (live-processor) kept a live consumer and stayed Stable. States were confirmed with kafka-consumer-groups.sh --list and --describe --state before and after each run.

Run	Inputs	Result
1	defaults, `min_idle_minutes: 0`	3 empty groups reported, `live-processor` not listed. Nothing deleted (`dry_run` true). `--list` unchanged
2	`auto_remediate: true`, `dry_run: true`	Same report, no `Pause`, nothing deleted. `--list` unchanged
3	`auto_remediate: true`, `dry_run: false`, `min_idle_minutes: 120`, fresh cursor	3 reported, all held as too recent, 0 past threshold, no `Pause`. `--list` unchanged
4	`auto_remediate: true`, `dry_run: false`, `min_idle_minutes: 0`, approved	Paused, resumed with `approved=true`. 3 groups deleted, re-list confirmed gone. `--list` shows only `live-processor`
5	same, resumed without approval	Paused, resumed with no answer. Nothing deleted. `--list` unchanged
6	approved, a consumer rejoined one target during the pause	Delete left that group behind, re-list still showed it, run failed and alerted
7	unreachable broker	Run failed, `errors` alert sent

Prerequisites

A Kafka cluster reachable from the Kestra worker, with a principal allowed to list and describe groups and, for remediation, delete them.
Stop a group's consumers before you expect it to be cleaned. A group with active members is not Empty, so it is never targeted.

Secrets

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
`bootstrap_servers`	`localhost:9092`	Brokers
`min_idle_minutes`	`1440`	How long a group must stay empty before it can be deleted
`protected_groups`	`[]`	Group ids never deleted
`auto_remediate`	`false`	Allow `ConsumerGroupDelete`
`dry_run`	`true`	Must be `false` for a delete to apply

Quick start

Add the SLACK_WEBHOOK_URL secret.
Import the flow and set bootstrap_servers.
Run it with the defaults and read the empty groups per cluster.
Set min_idle_minutes to how long a group must be abandoned before you trust the delete, for example a day.
Run it with auto_remediate: true and dry_run: false, then resume the paused execution with approved=true.
Enable the daily_scan trigger, which always runs in safe mode.

How to extend

Point the delete at a different cluster from the one that schedules the flow, by making bootstrap_servers the target cluster.
Archive each group's committed offsets to the KV store or object storage before deleting, so a wrongly cleaned group can be rebuilt.
Raise the Slack message to a channel with a link to the paused execution, so approval happens from chat.
Treat a group whose offsets are far behind the log end as a separate finding, since an empty group with large lag can mean a crashed consumer rather than a retired one.

Kafka Stale Consumer Group Cleanup Guard

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Kafka Stale Consumer Group Cleanup Guard

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Kafka Schema Registry Orphaned Subject Cleanup Guard

Kafka Schema Registry Compatibility and Mode Drift Guard

Real-Time Kafka Record Processing with Per-Record Executions

Kafka Schema Registry Subject Version Pruning Guard

Kafka Consumer Lag Sentinel

Block a breaking Avro schema before a Kafka producer deploys

More Infrastructure Blueprints

Nutanix AHV Inventory and Chargeback Report

Nomad Allocation Reschedule Storm Guard

Odoo Inventory Stock Threshold Gate

Nutanix safe patching with snapshot rollback

NPM Package Dependency Typosquatting Sentinel

Odoo Scheduled Action Overdue Sentinel

New to Kestra?