Kafka Idle Consumer Partition Guard with Kestra

id: kafka-idle-consumer-partition-guard
namespace: company.team

description: |
  Find Kafka consumer groups with more members than their topic has partitions: the extra
  consumers get no partition and sit idle while lag builds up on the others. For each listed
  group, compare active members with the partition count of its topics, alert Slack, and
  optionally raise the partition count to the number of members with TopicCreatePartitions,
  then describe the topic and the group again to verify every member has a partition.
  Compacted and keyed topics where ordering per key matters can be excluded.

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

  - id: groups
    type: ARRAY
    itemType: STRING
    displayName: Consumer groups to check
    defaults:
      - payments-processor
      - orders-sync

  - id: max_partitions
    type: INT
    displayName: Never grow a topic above (partitions)
    defaults: 48

  - id: protected_topics
    type: ARRAY
    itemType: STRING
    displayName: Topics never repartitioned
    description: Adding partitions changes which partition a key maps to, so per-key
      ordering breaks for keys already written. List keyed topics where that
      matters.
    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
      TopicCreatePartitions to run.
    defaults: true

tasks:
  - id: describe_groups
    type: io.kestra.plugin.kafka.ConsumerGroupDescribe
    description: Read-only. Members, their assigned partitions, and the topics each
      group reads.
    properties:
      bootstrap.servers: "{{ inputs.bootstrap_servers }}"
    groupIds: "{{ inputs.groups }}"

  - id: group_topics
    type: io.kestra.plugin.core.output.OutputValues
    values:
      topics: "{{ outputs.describe_groups.groups | jq('[.[].offsets[].topic] |
        unique') | first | toJson }}"

  - id: describe_topics
    type: io.kestra.plugin.core.flow.Loop
    values: "{{ outputs.group_topics.values.topics }}"
    outputs:
      - id: info
        type: STRING
        value: "{{ {'topic': item.value, 'partitions': outputs.describe.partitionCount,
          'cleanup': outputs.describe.configs['cleanup.policy']} | toJson }}"
    tasks:
      - id: describe
        type: io.kestra.plugin.kafka.TopicDescribe
        topic: "{{ item.value }}"
        properties:
          bootstrap.servers: "{{ inputs.bootstrap_servers }}"

  - id: findings
    type: io.kestra.plugin.core.output.OutputValues
    values:
      items: >-
        {{ {'g': outputs.describe_groups.groups, 't':
        outputs.describe_topics.outputs | jq('map(.outputs.info | fromjson |
        {(.topic): .}) | add // {}') | first, 'max': inputs.max_partitions,
        'prot': inputs.protected_topics} | jq('
           .t as $t | .max as $max | .prot as $prot
           | [.g[] | . as $g | ([.offsets[].topic] | unique) as $topics | ($g.members | length) as $m
              | $topics[] | . as $topic | $t[$topic] as $ti
              | ([$g.members[] | select((.assignedPartitions // []) | length == 0)] | length) as $idle
              | select($m > $ti.partitions)
              | {group: $g.groupId, topic: $topic, members: $m, partitions: $ti.partitions, idle_members: $idle, cleanup: $ti.cleanup,
                 target: ([$m, $max] | min),
                 action: (if ($prot | index($topic)) != null then "PROTECTED"
                          elif ($ti.cleanup | contains("compact")) then "COMPACTED"
                          elif $ti.partitions >= $max then "AT_MAX"
                          else "ADD_PARTITIONS" end)}]
           | group_by(.topic) | map(max_by(.members))') | first | toJson }}

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      Idle consumer check on {{ inputs.bootstrap_servers }} for {{ inputs.groups
      | join(', ') }}: {{ fromJson(outputs.findings.values.items) | length }}
      topics with more members than partitions. {{
      fromJson(outputs.findings.values.items) | jq('map("\(.group) on \(.topic):
      \(.members) members, \(.partitions) partitions, \(.idle_members) idle,
      \(if .action == "ADD_PARTITIONS" then "grow to \(.target)" else .action
      end)") | join("; ")') | first }}

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ fromJson(outputs.findings.values.items) | length > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': 'Kafka ' ~ inputs.bootstrap_servers ~ ': consumers sitting idle: ' ~ (fromJson(outputs.findings.values.items) | jq('map("\(.group) has \(.members) members for \(.partitions) partitions of \(.topic) (\(.idle_members) idle)") | join("; ")') | first) ~ '.' ~ (inputs.auto_remediate and not inputs.dry_run ? ' Adding partitions now.' : ' No change applied (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.findings.values.items) | jq('map(select(.action ==
          \"ADD_PARTITIONS\")) | length') | first > 0 }}"
        then:
          - id: remediate
            type: io.kestra.plugin.core.flow.Loop
            description: One TopicCreatePartitions per topic. The partition count can only
              grow.
            concurrencyLimit: 1
            values: "{{ fromJson(outputs.findings.values.items) | jq('map(select(.action ==
              \"ADD_PARTITIONS\"))') | first | toJson }}"
            tasks:
              - id: grow
                type: io.kestra.plugin.kafka.TopicCreatePartitions
                topic: "{{ fromJson(item.value).topic }}"
                totalPartitionCount: "{{ fromJson(item.value).target }}"
                properties:
                  bootstrap.servers: "{{ inputs.bootstrap_servers }}"

          - id: verify_topics
            type: io.kestra.plugin.core.flow.Loop
            values: "{{ fromJson(outputs.findings.values.items) | jq('map(select(.action ==
              \"ADD_PARTITIONS\") | .topic)') | first | toJson }}"
            outputs:
              - id: info
                type: STRING
                value: "{{ {'topic': item.value, 'partitions': outputs.recheck.partitionCount} |
                  toJson }}"
            tasks:
              - id: recheck
                type: io.kestra.plugin.kafka.TopicDescribe
                topic: "{{ item.value }}"
                properties:
                  bootstrap.servers: "{{ inputs.bootstrap_servers }}"

          - id: wait_rebalance
            type: io.kestra.plugin.core.flow.LoopUntil
            description: Consumers pick up new partitions at their next metadata refresh
              (metadata.max.age.ms, 5 minutes by default) and rebalance.
            condition: >-
              {{ {'g': outputs.regroup.groups ?? [], 'f':
              fromJson(outputs.findings.values.items)} | jq('[.f[] |
              select(.action == "ADD_PARTITIONS") | .group] as $gs
                 | [.g[] | select(.groupId as $x | $gs | index($x)) | .members[] | select((.assignedPartitions // []) | length == 0)] | length') | first == 0 }}
            checkFrequency:
              interval: PT20S
              maxIterations: 24
            tasks:
              - id: regroup
                type: io.kestra.plugin.kafka.ConsumerGroupDescribe
                properties:
                  bootstrap.servers: "{{ inputs.bootstrap_servers }}"
                groupIds: "{{ inputs.groups }}"

          - id: verify_result
            type: io.kestra.plugin.core.flow.If
            condition: >-
              {{ {'f': fromJson(outputs.findings.values.items), 'v':
              outputs.verify_topics.outputs | jq('map(.outputs.info | fromjson |
              {(.topic): .partitions}) | add') | first} | jq('.v as $v
                 | [.f[] | select(.action == "ADD_PARTITIONS") | select($v[.topic] != .target)] | length') | first > 0 }}
            then:
              - id: remediation_failed
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: "TopicCreatePartitions ran, but the partition count is not the
                  target on every topic."
            else:
              - id: remediated
                type: io.kestra.plugin.core.log.Log
                message: >-
                  Partitions added and verified: {{
                  fromJson(outputs.findings.values.items) |
                  jq('map(select(.action == "ADD_PARTITIONS") | "\(.topic)
                  \(.partitions) -> \(.target)") | join(", ")') | first }}. Idle
                  members after the rebalance: {{ {'g': outputs.regroup.groups}
                  | jq('[.g[].members[] | select((.assignedPartitions // []) |
                  length == 0)] | length') | first }}.
    else:
      - id: balanced
        type: io.kestra.plugin.core.log.Log
        message: "Every listed group has at least as many partitions as members."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Kafka idle consumer partition guard failed on ' ~ inputs.bootstrap_servers ~ ' (execution ' ~ execution.id ~ '). Check the logs.'} | toJson }}

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

Scaling a Kafka consumer from 2 to 5 replicas does nothing if the topic has 2 partitions. A partition is consumed by one member of a group at a time, so 3 consumers get no assignment and sit idle, while lag keeps growing on the other 2. The autoscaler, the dashboard and the pod count all say the service scaled.

This blueprint compares the active members of each consumer group with the partition count of the topics it reads. It alerts Slack with the number of idle members, and can raise the partition count to the number of members with TopicCreatePartitions. It then waits for the consumers to rebalance, and verifies that every member now has a partition.

This blueprint was created by zkasuran.

Decisions per topic

Action	Rule
`ADD_PARTITIONS`	More members than partitions: grow to the member count, capped at `max_partitions`
`PROTECTED`	The topic is in `protected_topics`: report only
`COMPACTED`	`cleanup.policy` includes `compact`: report only, since keys must stay on their partition
`AT_MAX`	Already at `max_partitions`: report only

Adding partitions is permanent, and it changes which partition a key maps to. Records with the same key written before and after the change can land on different partitions, so per-key ordering breaks across the change. List keyed topics where ordering matters in protected_topics.

How it works

describe_groups (io.kestra.plugin.kafka.ConsumerGroupDescribe, read-only) returns the members, their assigned partitions and the topics each group reads.
describe_topics (io.kestra.plugin.core.flow.Loop over io.kestra.plugin.kafka.TopicDescribe) reads the partition count and cleanup policy of each topic.
findings compares members with partitions per group and topic, counts the idle members and decides the action. summary logs it.
route (io.kestra.plugin.core.flow.If):
- balanced: logs it.
- findings: alert posts to Slack.
  - remediate_gate runs only when auto_remediate is true, dry_run is false, and a topic is ADD_PARTITIONS.
  - remediate (io.kestra.plugin.kafka.TopicCreatePartitions, one topic at a time) sets the new total.
  - verify_topics describes the topics again.
  - wait_rebalance (io.kestra.plugin.core.flow.LoopUntil) polls the groups every 20 seconds until no member is idle.
  - verify_result fails if a topic is not at its target count.
errors alerts Slack when the flow fails, for example on an unreachable broker.

What you get

The gap between how far a consumer scaled and how far it can actually parallelize, per group and topic.
Partition growth only where it is safe, with keyed, compacted and capped topics reported instead.
Proof after the change: the new partition count, and every member busy after the rebalance.

Who it's for

Teams that autoscale Kafka consumers on Kubernetes (KEDA, HPA) and see lag that scaling does not fix.
Platform teams running shared Kafka clusters with topics created at a few partitions.
Teams evaluating the Kafka plugin who want an example of ConsumerGroupDescribe driving TopicCreatePartitions.

Why orchestrate this with Kestra

The mismatch appears when a deployment scales, not when the topic is created. Kestra checks the groups on a schedule and sends the idle consumers to Slack. It grows the topic only behind auto_remediate and dry_run, and only where it is safe, waits for the rebalance with LoopUntil, verifies it, and keeps a history of every partition change.

Tested end to end

On Kestra 2.0.5 OSS against Apache Kafka 3.9.1 (KRaft), with a local HTTP sink standing in for Slack. Consumers were started with kafka-console-consumer.sh --group:

5 consumers in group payments-processor on the 2-partition topic payments-events;
1 consumer in orders-sync;
3 consumers in state-loader on the compacted 1-partition topic user-state.

Run	Inputs	Result
1	defaults	`payments-processor`: 5 members, 2 partitions, 3 idle, grow to 5. `orders-sync` not flagged. Alert sent
2	`payments-events` protected, remediate	Reported as `PROTECTED`. `kafka-topics.sh` still shows 2 partitions
3	dry run	Same report, nothing changed
4	remediate	2 to 5 partitions, verified. After the rebalance, `kafka-consumer-groups.sh --members` shows 1 partition for each of the 5 members
5	defaults	Balanced
6	`state-loader`, remediate	Reported as `COMPACTED`. `user-state` still has 1 partition
7	unreachable broker	`errors` alert sent

Prerequisites

A Kafka cluster reachable from the Kestra worker, and a principal allowed to describe groups and topics and, for remediation, alter topics.
Consumers pick up new partitions at their next metadata refresh (metadata.max.age.ms, 5 minutes by default). wait_rebalance waits up to 8 minutes.
Local testing: docker run -d --name kafka -p 9092:9092 apache/kafka:3.9.1, then create a 2-partition topic and start 5 kafka-console-consumer.sh with the same --group.

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
`groups`	2 example groups	Consumer groups to check
`max_partitions`	`48`	Upper bound when growing a topic
`protected_topics`	`[]`	Topics never repartitioned
`auto_remediate`	`false`	Allow `TopicCreatePartitions`
`dry_run`	`true`	Must be `false` for a change to apply

Quick start

Add the SLACK_WEBHOOK_URL secret.
Import the flow and set bootstrap_servers, groups, and protected_topics for keyed topics where per-key ordering matters.
Run it once and read the idle members per group.
Run it with auto_remediate: true and dry_run: false.
Enable the every_15_minutes trigger, which always runs in safe mode.

How to extend

Use the maximum replica count of the consumer's autoscaler instead of the current member count as the target.
Discover the groups with ConsumerGroupList instead of listing them.
Also flag the opposite case: few members and many partitions with growing lag, a sign the consumer should scale out.
Ask for approval with a Pause before growing topics read by several groups.

Kafka Idle Consumer Partition Guard

Decisions per topic

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 Idle Consumer Partition Guard

Decisions per topic

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 ACL Overly Broad Grant Audit

Fetch JSON from an API in Python and load it into MongoDB

Kafka Client Quota Guard

Kafka DLQ Quarantine, Classification, and Safe Replay Gate

Self-heal Kafka Connect connectors with a restart budget and escalation

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

More Infrastructure Blueprints

Detect Kubernetes CrashLoops and Post Triage Digests to Slack

IoT Edge Digital Twin with Drift Reconciliation

Cross-Stack Pipeline with Terraform, Ansible, and Jenkins

Kubernetes Deployment Restart Remediation from Alerts

Kubernetes Pod Lifecycle Management: Create, Inspect, and Cleanup

Jenkins Build-to-Deploy Approval Gate with Teams Notification

New to Kestra?