Kafka Schema Registry Orphaned Subject Cleanup Guard with Kestra

id: kafka-schema-registry-orphaned-subject-cleanup-guard
namespace: company.team

description: |
  Find Schema Registry subjects that no longer belong to a live Kafka topic, plus subjects already
  soft deleted but never hard deleted, then permanently delete only those behind an approval gate and
  re-list to confirm each one is gone. Under the TopicNameStrategy a registry keeps a `<topic>-value`
  and `<topic>-key` subject for every topic. When the topic is dropped its subjects stay, holding
  schema ids forever and cluttering compatibility checks. A subject whose topic was only just removed,
  where a recreate may be imminent, is reported but never deleted, because a first seen age is tracked
  per subject in a KV cursor so the clock has to run out first. A subject whose topic still exists is
  never a target.

inputs:
  - id: schema_registry_url
    type: STRING
    displayName: Schema Registry URL
    description: Base URL of the Confluent Schema Registry, with no trailing slash.
    defaults: http://localhost:8081

  - id: bootstrap_servers
    type: STRING
    displayName: Kafka bootstrap servers
    description: Comma separated broker list, used to read the live topic set.
    defaults: localhost:9092

  - id: min_age_minutes
    type: INT
    displayName: Minimum minutes a subject must stay orphaned before deletion
    description: An orphan seen for less than this is reported but never deleted, so
      a topic recreated inside the window keeps its subjects.
    defaults: 1440

  - id: hard_delete_soft_deleted
    type: BOOL
    displayName: Also hard delete subjects already soft deleted
    description: When true, subjects that are soft deleted but never permanently
      deleted are treated as orphans too.
    defaults: false

  - id: protected_subjects
    type: ARRAY
    itemType: STRING
    displayName: Subjects never deleted
    description: Subject names always kept, even when their topic is gone.
    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 any delete to run.
    defaults: true

variables:
  kv_key: "kafka_sr_orphans_{{ inputs.schema_registry_url | 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_age_minutes, 'MINUTES')
        | date("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'", timeZone="UTC") }}

  - id: list_subjects
    type: io.kestra.plugin.core.http.Request
    description: Read-only. Every registered subject the registry serves right now.
    method: GET
    uri: "{{ inputs.schema_registry_url }}/subjects"

  - id: list_all_subjects
    type: io.kestra.plugin.core.http.Request
    description: Read-only. Every subject including the ones already soft deleted.
    method: GET
    uri: "{{ inputs.schema_registry_url }}/subjects?deleted=true"

  - id: list_topics
    type: io.kestra.plugin.kafka.TopicList
    description: Read-only AdminClient call returning the live topics on the
      cluster, so a subject whose topic still exists is never flagged.
    properties:
      bootstrap.servers: "{{ inputs.bootstrap_servers }}"

  - id: analyze
    type: io.kestra.plugin.core.output.OutputValues
    description: Orphans are subjects following the topic name strategy whose topic
      is gone, plus the already soft deleted ones when that is enabled.
      Protected subjects are dropped.
    values:
      result: >-
        {{ {'live': fromJson(outputs.list_subjects.body), 'all':
        fromJson(outputs.list_all_subjects.body), 'topics':
        outputs.list_topics.topics, 'prot': inputs.protected_subjects,
        'hardsoft': inputs.hard_delete_soft_deleted} | jq('.live as $live | .all
        as $all | .topics as $topics | .prot as $prot | .hardsoft as $hs | ([
        $live[] | select(endswith("-value") or endswith("-key")) | . as $s | (if
        ($s|endswith("-value")) then ($s|rtrimstr("-value")) else
        ($s|rtrimstr("-key")) end) as $t | select(($topics | index($t)) | not) |
        select(([$s]-$prot)|length>0) | $s ]) as $missing | (if $hs then ([
        ($all - $live)[] | . as $s | select(([$s]-$prot)|length>0) ]) else []
        end) as $soft | {missing: $missing, soft: $soft, all: (($missing +
        $soft) | unique)}') | first | toJson }}

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ fromJson(outputs.analyze.values.result).all | length > 0 }}"
    then:
      - id: cursor_merge
        type: io.kestra.plugin.core.output.OutputValues
        description: Keep the first time each still orphaned subject was seen, add now
          for new ones, drop any subject no longer orphaned.
        values:
          merged: >-
            {{ {'cur': (kv(render(vars.kv_key), errorOnMissing=false) ?? "{}"),
            'ids': fromJson(outputs.analyze.values.result).all, 'now':
            outputs.freeze.values.now_iso} | jq('(.cur | if type=="string" then
            (if .=="" then {} else fromjson end) else . end) 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: split
        type: io.kestra.plugin.core.output.OutputValues
        values:
          target_ids: >-
            {{ {'cur': fromJson(outputs.cursor_merge.values.merged), 'ids':
            fromJson(outputs.analyze.values.result).all, '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.analyze.values.result).all, '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: >-
          Orphaned subject scan on {{ inputs.schema_registry_url }}: {{
          fromJson(outputs.analyze.values.result).missing | length }} subjects
          whose topic is gone ({{
          fromJson(outputs.analyze.values.result).missing | jq('join(", ")') |
          first }}), {{ fromJson(outputs.analyze.values.result).soft | length }}
          already soft deleted ({{ fromJson(outputs.analyze.values.result).soft
          | jq('join(", ")') | first }}). {{
          fromJson(outputs.split.values.target_ids) | length }} past the {{
          inputs.min_age_minutes }} minute age threshold ({{
          outputs.split.values.target_ids | jq('join(", ")') | first }}). {{
          fromJson(outputs.split.values.held_ids) | length }} held as too recent
          ({{ outputs.split.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': 'Schema Registry ' ~ inputs.schema_registry_url ~ ': ' ~
          (fromJson(outputs.analyze.values.result).all | length) ~ ' orphaned
          subjects, ' ~ (fromJson(outputs.split.values.target_ids) | length) ~ '
          past ' ~ inputs.min_age_minutes ~ ' minutes (' ~
          (outputs.split.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.split.values.target_ids) | length > 0) }}"
        then:
          - id: approve
            type: io.kestra.plugin.core.flow.Pause
            description: Resume with approved=true to permanently delete the subjects listed
              by log_findings. Resume without it or kill the execution to
              cancel.
            onResume:
              - id: approved
                type: BOOL
                displayName: Approve permanently deleting the orphaned subjects
                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: Remediation. Soft delete then permanent delete each targeted
                  subject, one at a time so the registry state never races.
                concurrencyLimit: 1
                values: "{{ outputs.split.values.target_ids }}"
                tasks:
                  - id: soft_delete
                    type: io.kestra.plugin.core.http.Request
                    description: Soft delete. A subject already soft deleted answers 404, which is
                      why the HTTP failure is allowed and the re-list is the
                      authority.
                    method: DELETE
                    uri: "{{ inputs.schema_registry_url }}/subjects/{{ item.value }}"
                    options:
                      allowFailed: true

                  - id: hard_delete
                    type: io.kestra.plugin.core.http.Request
                    description: Permanent delete, which frees the schema ids. allowFailed lets a
                      locked or read only subject fall through to the re-list
                      rather than aborting the loop.
                    method: DELETE
                    uri: "{{ inputs.schema_registry_url }}/subjects/{{ item.value }}?permanent=true"
                    options:
                      allowFailed: true

              - id: reverify
                type: io.kestra.plugin.core.http.Request
                description: Read-only re-list including soft deleted, so a subject that was not
                  fully removed still shows.
                method: GET
                uri: "{{ inputs.schema_registry_url }}/subjects?deleted=true"

              - id: verify_result
                type: io.kestra.plugin.core.flow.If
                condition: >-
                  {{ {'present': fromJson(outputs.reverify.body), 'targets':
                  fromJson(outputs.split.values.target_ids)} | jq('.present as
                  $p | [.targets[] | select(([.] - $p) | length == 0)] |
                  length') | first > 0 }}
                then:
                  - id: delete_failed
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: >-
                      Delete ran, but a re-list still shows a targeted subject.
                      Remaining: {{ {'present': fromJson(outputs.reverify.body),
                      'targets': fromJson(outputs.split.values.target_ids)} |
                      jq('.present as $p | [.targets[] | select(([.] - $p) |
                      length == 0)] | join(", ")') | first }}. The subject may
                      be read only or was re-registered during the pause.
                else:
                  - id: deleted_ok
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      Deleted {{ fromJson(outputs.split.values.target_ids) |
                      length }} orphaned subjects ({{
                      outputs.split.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 subject was deleted."
    else:
      - id: clear_cursor
        type: io.kestra.plugin.core.kv.Set
        description: No orphaned subjects, so reset the first seen cursor.
        key: "{{ render(vars.kv_key) }}"
        kvType: JSON
        value: "{}"

      - id: all_clean
        type: io.kestra.plugin.core.log.Log
        message: "No orphaned subjects on {{ inputs.schema_registry_url }}. Every
          registered subject maps to a live topic and nothing is left soft
          deleted. Nothing to clean."

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: >-
      {{ {'text': 'Kafka schema registry orphaned subject cleanup guard failed
      on ' ~ inputs.schema_registry_url ~ ' (execution ' ~ execution.id ~ '). No
      subject 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

A Confluent Schema Registry under the TopicNameStrategy keeps one subject per topic field: <topic>-value for the record value and <topic>-key for the key. The subjects are not tied to the topic lifecycle. Drop the topic and its subjects stay registered, holding their schema ids and version history. Over time a cluster collects subjects for topics nobody runs any more. They clutter GET /subjects, they confuse a compatibility check that reads the latest version of a dead subject and they pin schema ids that can never be reused.

Soft deletes leave the same debris. DELETE /subjects/<subject> without permanent=true only hides the subject from the default listing. The schema ids stay allocated until a second call with permanent=true runs. That second call is easy to forget.

This blueprint lists the registered subjects, lists the already soft deleted ones, reads the live Kafka topics and works out which subjects are orphaned. It tracks how long each has been orphaned in a KV cursor, so a subject deletes only after its topic has been gone past a threshold. Deletion is off by default, gated behind auto_remediate, dry_run and a human approval. After the delete it re-lists the subjects and fails the run if any targeted subject 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.
A subject whose topic still exists is never a target. The live topic set comes straight from the broker on every run.
The age cursor has no ttl, so a subject has to stay orphaned across the whole window before it is eligible. A topic dropped and recreated inside the window keeps its subjects.
protected_subjects are never deleted, even when their topic is gone.
Only subjects following the <topic>-value or <topic>-key naming are matched against topics. A custom named subject is never treated as topic bound.

How it works

freeze captures one timestamp for the run, so the detect, cursor and age steps all agree on now.
list_subjects (io.kestra.plugin.core.http.Request) reads GET /subjects, the subjects served right now.
list_all_subjects reads GET /subjects?deleted=true, which also returns the soft deleted ones. The soft deleted set is the difference between the two lists.
list_topics (io.kestra.plugin.kafka.TopicList, read-only) returns the live topics on the cluster.
analyze builds the orphan set. A subject is orphaned when it follows the topic name strategy and its topic is not in the live set or when hard_delete_soft_deleted is true and it is already soft deleted. Protected subjects are dropped.
route (io.kestra.plugin.core.flow.If) branches on whether any orphan exists.
- When none exist, clear_cursor resets the first seen record and all_clean logs that there is nothing to clean. No alert.
- When some exist:
  - cursor_merge and save_cursor keep the first time each still orphaned subject was seen in the KV key kafka_sr_orphans_<registry>, with no ttl.
  - split separates the orphans past min_age_minutes from the ones 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 subject is past the threshold.
    - approve (io.kestra.plugin.core.flow.Pause) waits for a resume carrying approved.
    - When approved, remediate_delete (io.kestra.plugin.core.flow.Loop, concurrencyLimit: 1) soft deletes then permanently deletes each targeted subject, reverify re-lists and verify_result fails the run if any targeted subject is still present.
    - When not approved, not_approved logs that nothing was deleted.
errors posts to Slack on any failure, for example an unreachable registry or a subject that survived the delete.

What you get

A clean inventory of orphaned subjects, split into topic gone and already soft deleted, separated from the subjects still in use.
An age gate that will not delete a subject whose topic was only just removed.
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 a shared registry where topics come and go faster than anyone prunes the subjects.
On-call engineers who want the dead subjects gone without running DELETE ...?permanent=true against production by hand.
Teams evaluating the Kafka plugin who want a worked example of TopicList paired with the registry REST API behind an approval gate.

Why orchestrate this with Kestra

A one line curl can delete a subject. It cannot remember how long the subject has been orphaned, cross check the live topic set, hold the delete for a second person or prove afterwards that the subject is really gone. Kestra adds the schedule that watches the registry, the KV cursor that measures the orphaned age across runs, the Pause that turns a permanent 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 Confluent Schema Registry 7.9.0 and a local HTTP sink standing in for Slack. Four topics (orders, payments, shipments, inventory) each had Avro subjects registered under the TopicNameStrategy, plus a non strategy subject my-custom-subject that must never be matched. State was confirmed with GET /subjects plus GET /subjects?deleted=true before and after each run.

Run	Inputs	Result
1	defaults, clean registry	0 orphans. All 6 subjects map to a live topic or are non strategy. Nothing deleted. No alert
2	topic `shipments` dropped, `payments-value` soft deleted, `min_age_minutes: 0`	2 orphans reported (`shipments-value` topic gone, `payments-value` soft deleted), both past threshold. `dry_run` true so nothing deleted. `?deleted=true` unchanged
3	same, `auto_remediate: true`, `dry_run: true`	Same report, no `Pause`, nothing deleted. Registry unchanged
4	`min_age_minutes: 120`, fresh cursor, `auto_remediate: true`, `dry_run: false`	2 orphans reported, both held as too recent, 0 past threshold, no `Pause`. Registry unchanged
5	`min_age_minutes: 0`, `auto_remediate: true`, `dry_run: false`, approved	Paused, resumed with `approved=true`. `shipments-value` soft then hard deleted, `payments-value` hard deleted. Re-list confirmed gone, `?deleted=true` dropped from 6 subjects to 4
6	topic `inventory` dropped, `inventory-value` set read only, approved	Delete rejected by read only mode, re-list still showed `inventory-value`, run failed and alerted
7	registry URL unreachable	`list_subjects` failed, run failed, `errors` alert sent

Across every run the subjects whose topic still exists (orders-value, orders-key) and the non strategy subject (my-custom-subject) were never flagged.

Prerequisites

A Kafka cluster plus a Confluent Schema Registry reachable from the Kestra worker. The registry keeps its state in a Kafka topic, so it cannot run without a broker. This is why the blueprint is not a demo.
A registry user allowed to list subjects and, for remediation, delete them. A broker principal allowed to list topics.
Subjects named with the default TopicNameStrategy (<topic>-value, <topic>-key). A different strategy needs the match in analyze adjusted.

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
`schema_registry_url`	`http://localhost:8081`	Registry base URL, no trailing slash
`bootstrap_servers`	`localhost:9092`	Brokers, read the live topic set
`min_age_minutes`	`1440`	How long a subject must stay orphaned before it can be deleted
`hard_delete_soft_deleted`	`false`	Also treat already soft deleted subjects as orphans
`protected_subjects`	`[]`	Subject names never deleted
`auto_remediate`	`false`	Allow the delete
`dry_run`	`true`	Must be `false` for a delete to apply

Quick start

Add the SLACK_WEBHOOK_URL secret.
Import the flow and set schema_registry_url plus bootstrap_servers.
Run it with the defaults and read the orphaned subjects per registry.
Set min_age_minutes to how long a topic must be gone 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 cleanup at a different registry from the one that schedules the flow, by making schema_registry_url the target registry.
Export each subject's schema versions to object storage before deleting, so a wrongly cleaned subject can be rebuilt.
Set hard_delete_soft_deleted to true to also reclaim the ids of subjects an operator soft deleted and forgot.
Match a different subject naming strategy by changing the topic extraction in analyze, for example RecordNameStrategy.
Raise the Slack message to a channel with a link to the paused execution, so approval happens from chat.

Kafka Schema Registry Orphaned Subject 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 Schema Registry Orphaned Subject 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 Compatibility and Mode Drift Guard

Kafka Schema Registry Subject Version Pruning Guard

Kafka DLQ Quarantine, Classification, and Safe Replay Gate

Kafka Orphaned Topic Reclaim Guard

Real-Time Kafka Record Processing with Per-Record Executions

Kafka Idle Consumer Partition Guard

More Infrastructure Blueprints

Nutanix safe patching with snapshot rollback

Odoo Scheduled Action Overdue Sentinel

Gate OIDC Signing-Key Rollovers with a Pinned JWKS

NPM Package Dependency Typosquatting Sentinel

Wave-Based VMware-to-AHV Migration Cutover

Gate Deployments on npm Audit Counts

New to Kestra?