Kafka Topic Disk Budget Guard with Kestra

id: kafka-topic-disk-budget-guard
namespace: company.team

description: |
  Keep Kafka topics inside a disk budget: read the on-disk size of every topic with
  DescribeLogDirs, compare it with a default budget and per-topic overrides, and flag topics
  over budget, especially those with no size or time limit at all. Alert Slack, and optionally
  set retention.bytes so each topic fits its budget (retention.bytes is per partition, so the
  budget is divided by the partition count), describe the topic again to verify, and optionally
  wait for the broker to delete the old segments. Compacted topics are only reported.

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

  - id: default_budget_mb
    type: INT
    displayName: Default budget per topic (MB)
    description: Total on-disk size allowed for a topic, all partitions and replicas
      included.
    defaults: 20

  - id: budgets_mb
    type: JSON
    displayName: Per-topic budgets (MB)
    defaults: '{"orders": 50}'

  - id: wait_for_reclaim
    type: BOOL
    displayName: Wait for the space to come back
    description: After remediation, poll DescribeLogDirs until the topics are within
      budget (the broker deletes closed segments on its own schedule).
    defaults: false

  - 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 TopicUpdate to run.
    defaults: true

tasks:
  - id: log_dirs
    type: io.kestra.plugin.kafka.DescribeLogDirs
    description: Read-only. On-disk size per topic, summed over partitions, replicas
      and brokers.
    properties:
      bootstrap.servers: "{{ inputs.bootstrap_servers }}"

  - id: over
    type: io.kestra.plugin.core.output.OutputValues
    values:
      topics: >-
        {{ {'sizes': outputs.log_dirs.topicSizes ?? {}, 'dflt':
        inputs.default_budget_mb, 'b': inputs.budgets_mb} | jq('
           [.sizes | to_entries[] | select(.key | startswith("__") | not)] as $s | .dflt as $dflt | .b as $b
           | [$s[] | {topic: .key, size: .value, budget: (($b[.key] // $dflt) * 1048576)} | select(.size > .budget)] | sort_by(-.size)') | first | toJson }}

  - id: describe_over
    type: io.kestra.plugin.core.flow.Loop
    description: Read the partition count and retention configs of every topic over budget.
    values: "{{ fromJson(outputs.over.values.topics) | jq('map(.topic)') | first |
      toJson }}"
    outputs:
      - id: info
        type: STRING
        value: >-
          {{ {'topic': item.value, 'partitions':
          outputs.describe.partitionCount, 'replicas':
          outputs.describe.replicationFactor, 'retention_ms':
          outputs.describe.configs['retention.ms'], 'retention_bytes':
          outputs.describe.configs['retention.bytes'], '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: plan
    type: io.kestra.plugin.core.output.OutputValues
    description: retention.bytes applies per partition and per replica, so the
      target is budget / (partitions x replication factor). A topic already
      capped at its target is left out, its overshoot is the active segments.
    values:
      items: >-
        {{ {'over': fromJson(outputs.over.values.topics), 'info':
        (outputs.describe_over.outputs ?? []) | jq('map(.outputs.info | fromjson
        | {(.topic): .}) | add // {}') | first} | jq('
           .info as $i | [.over[] | . + $i[.topic] | . + {
             unbounded: (.retention_ms == "-1" and .retention_bytes == "-1"),
             target_retention_bytes: ((.budget / (.partitions * .replicas)) | floor | tostring | sub("\\.0$"; "")),
             action: (if (.cleanup | contains("compact")) then "REPORT_COMPACTED" else "SET_RETENTION_BYTES" end)}
             | select(.action == "REPORT_COMPACTED" or .retention_bytes != .target_retention_bytes)]') | first | toJson }}

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      Kafka disk budget on {{ inputs.bootstrap_servers }}: {{
      outputs.log_dirs.topicSizes | length }} topics measured, {{
      fromJson(outputs.plan.values.items) | length }} over budget and not yet
      capped. {{ fromJson(outputs.plan.values.items) | jq('map("\(.topic)
      \((.size / 1048576 * 10 | floor) / 10) MB of \(.budget / 1048576 | floor |
      tostring | sub("\\.0$"; "")) MB\(if .unbounded then " (no time or size
      limit)" else "" end), \(.partitions) partitions: \(if .action ==
      "REPORT_COMPACTED" then "compacted, report only" else "retention.bytes ->
      \(.target_retention_bytes)" end)") | join("; ")') | first }}

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ fromJson(outputs.plan.values.items) | length > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': 'Kafka topics over their disk budget on ' ~ inputs.bootstrap_servers ~ ': ' ~ (fromJson(outputs.plan.values.items) | jq('map("\(.topic) \((.size / 1048576) | floor) MB / \(.budget / 1048576) MB\(if .unbounded then ", unbounded retention" else "" end)") | join("; ")') | first) ~ '.' ~ (inputs.auto_remediate and not inputs.dry_run ? ' Setting retention.bytes 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.plan.values.items) | jq('map(select(.action ==
          \"SET_RETENTION_BYTES\")) | length') | first > 0 }}"
        then:
          - id: remediate
            type: io.kestra.plugin.core.flow.Loop
            concurrencyLimit: 1
            values: "{{ fromJson(outputs.plan.values.items) | jq('map(select(.action ==
              \"SET_RETENTION_BYTES\"))') | first | toJson }}"
            tasks:
              - id: update
                type: io.kestra.plugin.kafka.TopicUpdate
                topic: "{{ fromJson(item.value).topic }}"
                configs:
                  retention.bytes: "{{ fromJson(item.value).target_retention_bytes }}"
                properties:
                  bootstrap.servers: "{{ inputs.bootstrap_servers }}"

          - id: verify
            type: io.kestra.plugin.core.flow.Loop
            values: "{{ fromJson(outputs.plan.values.items) | jq('map(select(.action ==
              \"SET_RETENTION_BYTES\") | .topic)') | first | toJson }}"
            outputs:
              - id: live
                type: STRING
                value: "{{ {'topic': item.value, 'retention_bytes':
                  outputs.recheck.configs['retention.bytes']} | toJson }}"
            tasks:
              - id: recheck
                type: io.kestra.plugin.kafka.TopicDescribe
                topic: "{{ item.value }}"
                properties:
                  bootstrap.servers: "{{ inputs.bootstrap_servers }}"

          - id: verify_result
            type: io.kestra.plugin.core.flow.If
            condition: >-
              {{ {'p': fromJson(outputs.plan.values.items), 'l':
              outputs.verify.outputs} | jq('(.l | map(.outputs.live | fromjson |
              {(.topic): .retention_bytes}) | add) as $l
                 | [.p[] | select(.action == "SET_RETENTION_BYTES") | select($l[.topic] != .target_retention_bytes)] | length') | first > 0 }}
            then:
              - id: remediation_failed
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: "TopicUpdate ran, but retention.bytes is not the planned value on
                  every topic."
            else:
              - id: remediated
                type: io.kestra.plugin.core.log.Log
                message: "retention.bytes set and verified on {{
                  fromJson(outputs.plan.values.items) | jq('map(select(.action
                  == \"SET_RETENTION_BYTES\") |
                  \"\\(.topic)=\\(.target_retention_bytes)\") | join(\", \")') |
                  first }}."

          - id: wait_reclaim
            type: io.kestra.plugin.core.flow.LoopUntil
            runIf: "{{ inputs.wait_for_reclaim }}"
            description: The broker deletes closed segments at its next retention check. The
              active segment always stays, so up to one segment per partition
              above the budget is accepted.
            condition: >-
              {{ {'s': outputs.measure.topicSizes ?? {}, 'p':
              fromJson(outputs.plan.values.items)} | jq('.s as $s | [.p[] |
              select(.action == "SET_RETENTION_BYTES") | select(($s[.topic] //
              0) > .budget * 1.25)] | length') | first == 0 }}
            checkFrequency:
              interval: PT15S
              maxIterations: 40
            tasks:
              - id: measure
                type: io.kestra.plugin.kafka.DescribeLogDirs
                properties:
                  bootstrap.servers: "{{ inputs.bootstrap_servers }}"

          - id: reclaimed
            type: io.kestra.plugin.core.log.Log
            runIf: "{{ inputs.wait_for_reclaim }}"
            message: >-
              Space reclaimed: {{ {'s': outputs.measure.topicSizes, 'p':
              fromJson(outputs.plan.values.items)} | jq('.s as $s | [.p[] |
              select(.action == "SET_RETENTION_BYTES") | "\(.topic) \((.size /
              1048576) | floor | tostring | sub("\\.0$"; "")) MB ->
              \(($s[.topic] / 1048576 * 10 | floor) / 10) MB"] | join(", ")') |
              first }}.
    else:
      - id: within_budget
        type: io.kestra.plugin.core.log.Log
        message: "Every topic is within its disk budget, or already capped at it."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Kafka topic disk budget guard failed on ' ~ inputs.bootstrap_servers ~ ' (execution ' ~ execution.id ~ '). Disk usage unknown, check the logs.'} | toJson }}

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

Kafka brokers usually run out of disk because of one topic, not the total. Typical causes:

a clickstream topic created with retention.ms=-1 "for now";
a topic whose traffic grew tenfold;
a debugging topic nobody owns.

Time-based retention does not help when the volume per day changes, and nothing warns before the disk alarm.

This blueprint reads the on-disk size of every topic with the AdminClient and compares it with a budget. It alerts Slack, and can cap the topics over budget with retention.bytes, then describe them again to verify. Optionally it waits until the broker has deleted the old segments and the space is back.

This blueprint was created by zkasuran.

The budget

default_budget_mb applies to every topic, and budgets_mb overrides it per topic. Internal topics (__consumer_offsets...) are skipped.
A budget is the total on disk: all partitions, all replicas, all brokers, as returned by DescribeLogDirs.
retention.bytes is enforced per partition replica, so the value set is budget / (partitions × replication factor). A 20 MB budget on a 4-partition topic sets retention.bytes=5242880.
Compacted topics are reported but never capped: a size limit would delete live keys.
A topic already capped at its target is not reported again. Its overshoot is the active segments, which Kafka never deletes.

How it works

log_dirs (io.kestra.plugin.kafka.DescribeLogDirs, read-only) returns the size of every topic. over keeps those above their budget.
describe_over (io.kestra.plugin.core.flow.Loop over io.kestra.plugin.kafka.TopicDescribe) reads the partition count, replication factor and retention configs of each one.
plan computes the per-partition retention.bytes, flags unbounded topics (retention.ms and retention.bytes both -1), and marks compacted ones as report only. summary logs it.
route (io.kestra.plugin.core.flow.If):
- within budget: logs it.
- over budget: alert posts to Slack.
  - remediate_gate runs only when auto_remediate is true and dry_run is false.
  - remediate sets retention.bytes with TopicUpdate, then verify and verify_result describe the topics again and fail on any other value.
  - With wait_for_reclaim, wait_reclaim (io.kestra.plugin.core.flow.LoopUntil) polls DescribeLogDirs every 15 seconds until each topic is within 125% of its budget, then reclaimed logs the sizes before and after.
errors alerts Slack when the flow fails, for example on an unreachable broker.

What you get

Disk usage per topic, measured by the brokers, against a budget you set.
The retention.bytes arithmetic done for you, per partition and replica.
Proof that the cap is set, and optionally that the space really came back.

Who it's for

Platform teams running Kafka for many producers on fixed-size disks.
SREs who get disk alarms on brokers and want to know which topic caused them.
Teams evaluating the Kafka plugin who want an example of DescribeLogDirs driving TopicUpdate.

Why orchestrate this with Kestra

kafka-log-dirs.sh prints raw JSON per partition. Kestra sums it per topic on a schedule and compares it with budgets kept next to the flow. It turns overruns into a Slack alert, applies the cap only behind auto_remediate and dry_run, waits for the broker with LoopUntil, and keeps an execution history of every topic's size over time.

Tested end to end

On Kestra 2.0.5 OSS against Apache Kafka 3.9.1 (KRaft, log.retention.check.interval.ms=10000, 1 MB segments), with a local HTTP sink standing in for Slack. Topics:

clickstream: 4 partitions, 58 MB, retention.ms=-1;
orders: 4 MB, with a 50 MB budget;
customers: compacted, 1.8 MB.

Run	Inputs	Result
1	defaults (20 MB)	`clickstream` over budget with no limit, target `retention.bytes=5242880`. Alert sent
2	`default_budget_mb: 1`	Adds `customers`, reported as compacted, never capped. `orders` within its 50 MB override
3	`auto_remediate: true`, `dry_run: true`	Same report. `kafka-configs.sh` shows no `retention.bytes` on `clickstream`
4	`auto_remediate: true`, `dry_run: false`, `wait_for_reclaim: true`	`retention.bytes=5242880` set and verified. `kafka-configs.sh` shows the same. `clickstream` went from 57 MB to 22.3 MB on disk
5	defaults again	Nothing to do: `clickstream` is capped, the 2.3 MB above budget is the active segments
6	unreachable broker	`errors` alert sent

Prerequisites

A Kafka cluster reachable from the Kestra worker, and a principal allowed to describe the cluster and topics and, for remediation, alter topic configs.
Local testing:
- docker run -d --name kafka -p 9092:9092 apache/kafka:3.9.1;
- create a topic with --config retention.ms=-1 --config segment.bytes=1048576;
- fill it with kafka-producer-perf-test.sh.

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
`default_budget_mb`	`20`	Budget per topic
`budgets_mb`	`{"orders": 50}`	Per-topic overrides
`wait_for_reclaim`	`false`	Wait for the space to come back
`auto_remediate`	`false`	Allow `TopicUpdate`
`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, default_budget_mb and budgets_mb for your largest topics.
Run it once and read the topics over budget.
Check with the owners that the data beyond the budget can go, then run with auto_remediate: true and dry_run: false.
Enable the hourly trigger, which always runs in safe mode.

How to extend

Alert at 80% of the budget, before the topic is over.
Load budgets from a namespace file or Git, owned by the teams that produce to each topic.
Track the size per topic in a table to forecast when a broker disk will be full.
Ask for approval with a Pause before capping topics that hold more than a day of data.

Kafka Topic Disk Budget Guard

The budget

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 Topic Disk Budget Guard

The budget

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

Write-Audit-Publish (WAP) and Atomic Merge Gate for Apache Iceberg with Project Nessie

NocoDB Database Platform Health, Storage Engine, and Project Sentinel

OceanBase Major Compaction Stall & Auto-Resume Sentinel

Search OpenAlex for Related Scholarly Works

Automated OpenLineage Metadata Extraction and Marquez Governance Pipeline

OpenMetadata Catalog Health and Change-Event Sentinel

More Infrastructure Blueprints

CircleCI Pipeline Failure and Flake Ratchet Gate

Score change requests on risk and build the CAB agenda, pre-approving the low-risk ones

Chrony NTP Stratum Dispersion and Clock Drift Sentinel

ChaosBlade Experiment Fault Injection State Sentinel

CI/CD Failure Diagnosis for GitHub Actions and Kestra

Kubernetes Chaos Game Day with Parallel Blast-Radius Watch

New to Kestra?