Kafka Orphaned Topic Reclaim Guard

id: kafka-orphaned-topic-reclaim-guard
namespace: company.team

description: |
  Find Kafka topics that nothing reads and nothing has committed an offset against, hold them
  until they have looked abandoned for a whole grace window, then delete only those behind an
  approval gate and re-list to prove each one is gone. A decommissioned service leaves its topics
  on the cluster. A producer typo plus auto-create leaves a topic nobody asked for. Both hold
  partitions, replicas and disk for good, because nothing in a normal Kafka workflow removes a
  topic. Internal topics, topics with a committed consumer group offset, protected topics and
  topics whose size or layout changed since the last run are never touched.

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

  - id: min_idle_minutes
    type: INT
    displayName: Minutes a topic must look abandoned before it can be deleted
    description: Measured from the first run that saw the topic unused with its
      current size and layout. A topic seen for less than this is reported but
      never deleted.
    defaults: 10080

  - id: protected_topics
    type: ARRAY
    itemType: STRING
    displayName: Topics never deleted
    description: Topic names always kept even when nothing reads them, for example a
      landing topic that a batch job drains once a quarter.
    defaults: []

  - id: verify_delay
    type: DURATION
    displayName: Wait before the re-list
    description: Kafka marks a topic for deletion and removes it shortly after, so
      the re-list waits this long before deciding whether the delete landed.
    defaults: PT5S

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

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

tasks:
  - id: freeze
    type: io.kestra.plugin.core.output.OutputValues
    description: One timestamp for the whole run, so the detect, the cursor and the
      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.TopicList
    description: Read-only. Every topic on the cluster, internal ones included, so
      the count of what was excluded is visible in the report.
    includeInternal: true
    properties:
      bootstrap.servers: "{{ inputs.bootstrap_servers }}"

  - id: list_groups
    type: io.kestra.plugin.kafka.ConsumerGroupList
    description: Read-only. Every consumer group, whatever its state, because a
      group that is Empty today still holds the offsets that prove its topic is
      in use.
    properties:
      bootstrap.servers: "{{ inputs.bootstrap_servers }}"

  - id: log_dirs
    type: io.kestra.plugin.kafka.DescribeLogDirs
    description: Read-only. Bytes on disk per topic. Used both to report what the
      orphans cost and to notice a topic that is still being written to.
    properties:
      bootstrap.servers: "{{ inputs.bootstrap_servers }}"

  - id: groups_present
    type: io.kestra.plugin.core.flow.If
    description: ConsumerGroupDescribe rejects an empty groupIds list, so a cluster
      with no groups skips the describe.
    condition: "{{ outputs.list_groups.groups | length > 0 }}"
    then:
      - id: describe_groups
        type: io.kestra.plugin.kafka.ConsumerGroupDescribe
        description: Read-only. Members and committed offsets of every group, which is
          what marks a topic as in use.
        groupIds: "{{ outputs.list_groups.groups | jq('[.[].groupId]') | first }}"
        properties:
          bootstrap.servers: "{{ inputs.bootstrap_servers }}"

  - id: analyze
    type: io.kestra.plugin.core.output.OutputValues
    description: >-
      A topic is a candidate only when it survives every exclusion: not
      internal, no committed offset from any group, not assigned to a live
      member, not on the protect list.
    values:
      result: >-
        {{ {'all': outputs.list_all.topics, 'g': ((outputs.describe_groups is
        defined) ? outputs.describe_groups.groups : []), 'prot':
        inputs.protected_topics, 'sizes': (outputs.log_dirs.topicSizes ?? {})} |
        jq('.all as $all | .g as $g | .prot as $prot | .sizes as $sizes |
        ([$all[] | select(startswith("_"))] | sort) as $internal | (([$g[] |
        (.offsets // [])[] | .topic] + [$g[] | (.members // [])[] |
        (.assignedPartitions // [])[] | sub("-[0-9]+$";"")]) | unique) as
        $consumed | ([$all[] | select(startswith("_") | not) | select(([.] -
        $consumed) | length > 0) | select(([.] - $prot) | length > 0)] | sort)
        as $cand | {internal: $internal, consumed: ([$consumed[] |
        select(startswith("_") | not)] | sort), protected_present: ([$all[] |
        select(([.] - $prot) | length == 0)] | sort), candidates: $cand,
        orphan_bytes: ([$cand[] | ($sizes[.] // 0)] | add // 0)}') | first |
        toJson }}

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ fromJson(outputs.analyze.values.result).candidates | length > 0 }}"
    then:
      - id: describe_candidates
        type: io.kestra.plugin.core.flow.Loop
        description: Read-only. Partition count and replication factor of each
          candidate, which is what the topic is actually holding.
        values: "{{ fromJson(outputs.analyze.values.result).candidates | toJson }}"
        concurrencyLimit: 4
        outputs:
          - id: info
            type: STRING
            value: >-
              {{ {'topic': item.value, 'partitions':
              outputs.describe_one.partitionCount, 'replicas':
              outputs.describe_one.replicationFactor} | toJson }}
        tasks:
          - id: describe_one
            type: io.kestra.plugin.kafka.TopicDescribe
            topic: "{{ item.value }}"
            properties:
              bootstrap.servers: "{{ inputs.bootstrap_servers }}"

      - id: fingerprint
        type: io.kestra.plugin.core.output.OutputValues
        description: The state a topic has to hold still in. Any change to bytes,
          partitions or replicas restarts its grace window.
        values:
          now_map: >-
            {{ {'l': outputs.describe_candidates.outputs, 'sizes':
            (outputs.log_dirs.topicSizes ?? {})} | jq('.sizes as $s | ((.l |
            map(.outputs.info | fromjson) | map({(.topic): {size: ($s[.topic] //
            0), partitions: .partitions, replicas: .replicas}}) | add) // {})')
            | first | toJson }}

      - id: cursor_merge
        type: io.kestra.plugin.core.output.OutputValues
        description: Keep the first time each candidate was seen in its current state. A
          candidate that is new gets now. So does one whose state moved. A topic
          that is no longer a candidate is dropped.
        values:
          merged: >-
            {{ {'cur': (kv(render(vars.kv_key), errorOnMissing=false) ?? ''),
            'f': fromJson(outputs.fingerprint.values.now_map), 'now':
            outputs.freeze.values.now_iso} | jq('(.cur | if type=="string" then
            (if . == "" then {} else fromjson end) else . end) as $c | .now as
            $n | .f as $f | reduce ($f | keys_unsorted[]) as $t ({}; . + {($t):
            (if ($c[$t] != null and $c[$t].size == $f[$t].size and
            $c[$t].partitions == $f[$t].partitions and $c[$t].replicas ==
            $f[$t].replicas) then $c[$t] else ($f[$t] + {first_seen: $n})
            end)})') | first | toJson }}

      - id: save_cursor
        type: io.kestra.plugin.core.kv.Set
        description: No ttl. This 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: split
        type: io.kestra.plugin.core.output.OutputValues
        values:
          target_ids: >-
            {{ {'cur': fromJson(outputs.cursor_merge.values.merged), 'cutoff':
            outputs.freeze.values.cutoff_iso} | jq('.cutoff as $cut | [.cur |
            to_entries[] | select(.value.first_seen <= $cut) | .key] | sort') |
            first | toJson }}
          held_ids: >-
            {{ {'cur': fromJson(outputs.cursor_merge.values.merged), 'cutoff':
            outputs.freeze.values.cutoff_iso} | jq('.cutoff as $cut | [.cur |
            to_entries[] | select(.value.first_seen > $cut) | .key] | sort') |
            first | toJson }}
          target_bytes: >-
            {{ {'cur': fromJson(outputs.cursor_merge.values.merged), 'cutoff':
            outputs.freeze.values.cutoff_iso} | jq('.cutoff as $cut | ([.cur |
            to_entries[] | select(.value.first_seen <= $cut) | .value.size] |
            add // 0)') | first }}
          detail: >-
            {{ {'cur': fromJson(outputs.cursor_merge.values.merged)} | jq('[.cur
            | to_entries[] | "\(.key): \(.value.partitions) partitions x
            \(.value.replicas) replicas, \(.value.size) bytes, unused since
            \(.value.first_seen)"] | join(" | ")') | first }}

      - id: log_findings
        type: io.kestra.plugin.core.log.Log
        message: >-
          Orphaned topic scan on {{ inputs.bootstrap_servers }}: {{
          outputs.list_all.topics | length }} topics on the cluster, {{
          fromJson(outputs.analyze.values.result).internal | length }} internal
          skipped, {{ fromJson(outputs.analyze.values.result).consumed | length
          }} in use by a consumer group, {{
          fromJson(outputs.analyze.values.result).protected_present | length }}
          on the protect list. {{
          fromJson(outputs.analyze.values.result).candidates | length }} unused
          ({{ outputs.analyze.values.result | jq('.candidates | join(", ")') |
          first }}) holding {{
          fromJson(outputs.analyze.values.result).orphan_bytes }} bytes. {{
          fromJson(outputs.split.values.target_ids) | length }} past the {{
          inputs.min_idle_minutes }} minute grace window ({{
          outputs.split.values.target_ids | jq('join(", ")') | first }}) holding
          {{ outputs.split.values.target_bytes }} bytes. {{
          fromJson(outputs.split.values.held_ids) | length }} held as too recent
          or recently changed ({{ outputs.split.values.held_ids | jq('join(",
          ")') | first }}). Detail: {{ outputs.split.values.detail }}.
          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.analyze.values.result).candidates | length) ~ '
          orphaned topics holding ' ~
          fromJson(outputs.analyze.values.result).orphan_bytes ~ ' bytes, ' ~
          (fromJson(outputs.split.values.target_ids) | length) ~ ' past the ' ~
          inputs.min_idle_minutes ~ ' minute grace window (' ~
          (outputs.split.values.target_ids | jq('join(", ")') | first) ~ ').
          Deleting a topic is irreversible. 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.split.values.target_ids) | length > 0) }}"
        then:
          - id: approve
            type: io.kestra.plugin.core.flow.Pause
            description: >-
              Deleting a Kafka topic is irreversible. Its log segments, its
              partitions and every message still inside them go for good. No
              Kafka command brings them back. Read the topic list in
              log_findings first, then resume with approved=true to delete
              exactly those topics. Resume with no answer or kill the execution
              to cancel.
            onResume:
              - id: approved
                type: BOOL
                displayName: Approve the irreversible delete of the listed orphaned topics
                required: false

          - id: do_delete
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.approve.onResume.approved ?? false }}"
            then:
              - id: remediate_delete
                type: io.kestra.plugin.core.flow.Loop
                description: One topic per iteration and one iteration at a time, so a refusal
                  is attributable to a single topic instead of sinking the
                  batch.
                values: "{{ outputs.split.values.target_ids }}"
                concurrencyLimit: 1
                tasks:
                  - id: delete_topic
                    type: io.kestra.plugin.kafka.TopicDelete
                    description: allowFailure keeps the re-list as the authority on what actually
                      went.
                    allowFailure: true
                    topics:
                      - "{{ item.value }}"
                    properties:
                      bootstrap.servers: "{{ inputs.bootstrap_servers }}"

              - id: settle
                type: io.kestra.plugin.core.flow.Sleep
                description: Kafka marks a topic for deletion, then removes it. Give the
                  controller that long before reading back.
                duration: "{{ inputs.verify_delay }}"

              - id: reverify
                type: io.kestra.plugin.kafka.TopicList
                description: Read-only re-list after the delete. This is the gate.
                includeInternal: true
                properties:
                  bootstrap.servers: "{{ inputs.bootstrap_servers }}"

              - id: verify_result
                type: io.kestra.plugin.core.flow.If
                condition: >-
                  {{ {'now': outputs.reverify.topics, 'targets':
                  fromJson(outputs.split.values.target_ids)} | jq('[.now[]] as
                  $present | [.targets[] | select(([.] - $present) | length ==
                  0)] | length') | first > 0 }}
                then:
                  - id: delete_failed
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: >-
                      TopicDelete ran, but a re-list still shows a targeted
                      topic. Remaining: {{ {'now': outputs.reverify.topics,
                      'targets': fromJson(outputs.split.values.target_ids)} |
                      jq('[.now[]] as $present | [.targets[] | select(([.] -
                      $present) | length == 0)] | join(", ")') | first }}.
                      Either the delete was refused (delete.topic.enable false
                      on the broker, the principal missing Delete on the topic)
                      or a client wrote to the topic and auto-create brought it
                      straight back, which means it was not abandoned after all.
                else:
                  - id: prune_cursor
                    type: io.kestra.plugin.core.kv.Set
                    description: Drop the deleted topics from the first-seen record, so a topic
                      later recreated under the same name starts a fresh grace
                      window instead of inheriting the old one.
                    key: "{{ render(vars.kv_key) }}"
                    kvType: JSON
                    value: >-
                      {{ {'cur': fromJson(outputs.cursor_merge.values.merged),
                      'gone': fromJson(outputs.split.values.target_ids)} |
                      jq('.gone as $g | (.cur | with_entries(select(([.key] -
                      $g) | length > 0)))') | first | toJson }}

                  - id: deleted_ok
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      Deleted {{ fromJson(outputs.split.values.target_ids) |
                      length }} orphaned topics ({{
                      outputs.split.values.target_ids | jq('join(", ")') | first
                      }}) and reclaimed {{ outputs.split.values.target_bytes }}
                      bytes of log data. A re-list confirms none of them remain.
                      Cluster now holds {{ outputs.reverify.topics | length }}
                      topics.
            else:
              - id: not_approved
                type: io.kestra.plugin.core.log.Log
                message: "Pause resumed without approval or killed. No topic was deleted."
    else:
      - id: clear_cursor
        type: io.kestra.plugin.core.kv.Set
        description: Nothing unused, so reset the first-seen record.
        key: "{{ render(vars.kv_key) }}"
        kvType: JSON
        value: "{}"

      - id: all_in_use
        type: io.kestra.plugin.core.log.Log
        message: >-
          Every topic on {{ inputs.bootstrap_servers }} is internal, in use by a
          consumer group or on the protect list. {{ outputs.list_all.topics |
          length }} topics checked, nothing to reclaim.

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

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