Reclaim Pulsar ledger storage from orphaned subscriptions with Kestra

id: pulsar-orphaned-subscription-backlog-clear-guard
namespace: company.team

description: >-
  Find Apache Pulsar subscriptions that have no consumers but still pin a large
  message backlog, then clear or remove them behind a dry run, an approval
  pause, a live re-check and a re-read that fails the run.

inputs:
  - id: admin_url
    type: STRING
    displayName: Pulsar admin REST base URL
    defaults: http://pulsar-broker:8080

  - id: broker_uri
    type: STRING
    displayName: Pulsar broker service URL
    defaults: pulsar://pulsar-broker:6650

  - id: auth_mode
    type: SELECT
    displayName: Admin API authentication mode
    values:
      - token
      - none
    defaults: token

  - id: tenant
    type: STRING
    displayName: Pulsar tenant
    defaults: public

  - id: pulsar_namespace
    type: STRING
    displayName: Pulsar namespace inside the tenant
    defaults: default

  - id: min_backlog_messages
    type: INT
    displayName: Flag a consumerless subscription at or above this backlog
    defaults: 100000

  - id: grace_period_hours
    type: INT
    displayName: Hours a subscription must stay consumerless before it can be touched
    defaults: 72

  - id: target_backlog_messages
    type: INT
    displayName: Backlog the re-read must show after a skip_all
    defaults: 0

  - id: remediation_action
    type: SELECT
    displayName: What to do with an orphaned subscription
    values:
      - skip_all
      - unsubscribe
    defaults: skip_all

  - id: max_subscriptions
    type: INT
    displayName: Most subscriptions to act on in one run
    defaults: 5

  - id: sample_messages
    type: INT
    displayName: Messages to read from each target topic as evidence before the write
    defaults: 5

  - id: dry_run
    type: BOOL
    displayName: Report the plan without calling any admin write
    defaults: true

  - id: auto_remediate
    type: BOOL
    displayName: Allow the admin write after approval
    defaults: false

variables:
  first_seen_key: pulsar_orphan_subs_first_seen
  inventory_program: >-
    (.stats | fromjson) as $s | .ns as $n | (.topics | fromjson) as $live | [
    ($s[$n] // {}) | to_entries[] | .value | to_entries[] | .value |
    to_entries[] | .key as $t | .value.subscriptions as $subs | ($subs // {}) |
    to_entries[] | {topic: $t, sub: .key, backlog: (.value.msgBacklog // 0),
    consumers: ((.value.consumers // []) | length), kind: ($t | split("://") |
    .[0]), path: ($t | split("://") | .[1])} ] | map(select((.topic |
    contains("/__")) == false and (.sub | startswith("__")) == false)) |
    map(select((([.topic] - $live) | length) == 0))
  subscription_program: >-
    (.body | fromjson) as $d | .sub as $n | (($d.subscriptions // {})[$n]) as $r
    | if $r == null then {present: false, backlog: -1, bytes: -1, consumers: -1,
    last_consumed: "none"} else {present: true, backlog: ($r.msgBacklog // 0),
    bytes: ($r.backlogSize // 0), consumers: ((($r.consumers) // []) | length),
    last_consumed: (($r.lastConsumedTimestamp // 0) | tostring)} end

tasks:
  - id: list_topics
    type: io.kestra.plugin.core.http.Request
    description: Ask the broker which persistent topics the namespace holds right now.
    method: GET
    uri: "{{ inputs.admin_url }}/admin/v2/persistent/{{ inputs.tenant }}/{{
      inputs.pulsar_namespace }}"
    headers:
      Authorization: "{{ inputs.auth_mode == 'token' ? 'Bearer ' ~
        secret('PULSAR_ADMIN_TOKEN') : 'Bearer none' }}"

  - id: read_broker_stats
    type: io.kestra.plugin.core.http.Request
    description: One call returns every topic the broker owns with its
      subscriptions, backlogs and consumers.
    method: GET
    uri: "{{ inputs.admin_url }}/admin/v2/broker-stats/topics"
    headers:
      Authorization: "{{ inputs.auth_mode == 'token' ? 'Bearer ' ~
        secret('PULSAR_ADMIN_TOKEN') : 'Bearer none' }}"

  - id: inventory
    type: io.kestra.plugin.core.output.OutputValues
    description: Reduce the stats tree to one row per subscription on a topic that
      still exists.
    values:
      subscriptions: "{{ {'stats': outputs.read_broker_stats.body, 'ns': inputs.tenant
        ~ '/' ~ inputs.pulsar_namespace, 'topics': outputs.list_topics.body} |
        toJson | jq(render(vars.inventory_program)) | first | toJson }}"

  - id: read_first_seen
    type: io.kestra.plugin.core.output.OutputValues
    description: Load the first-seen snapshot that drives the age gate.
    values:
      snapshot: "{{ kv(render(vars.first_seen_key), errorOnMissing=false) ?? '' }}"

  - id: build_plan
    type: io.kestra.plugin.core.output.OutputValues
    description: Split the consumerless backlog holders into aged targets and in-grace ones.
    values:
      plan: "{{ {'subs': outputs.inventory.values.subscriptions, 'seen':
        outputs.read_first_seen.values.snapshot, 'now': execution.startDate |
        date(\"yyyy-MM-dd'T'HH:mm:ss'Z'\", 'UTC'), 'cutoff': execution.startDate
        | dateAdd(-1 * inputs.grace_period_hours, 'HOURS') |
        date(\"yyyy-MM-dd'T'HH:mm:ss'Z'\", 'UTC'), 'thr':
        inputs.min_backlog_messages, 'max': inputs.max_subscriptions, 'act':
        inputs.remediation_action} | toJson | jq('(.subs | fromjson) as $subs |
        (if (.seen | type) == \"string\" then (if .seen == \"\" then {} else
        (.seen | fromjson) end) else (.seen // {}) end) as $seen | .now as $now
        | .cutoff as $cutoff | (.thr | tonumber) as $thr | (.max | tonumber) as
        $max | .act as $act | [ $subs[] | select(.consumers == 0 and .backlog >=
        $thr) | . + {key: (.topic + \"|\" + .sub)} ] | [ .[] | . + {first_seen:
        ($seen[.key] // $now)} ] | [ .[] | . + {action_path: (if $act ==
        \"unsubscribe\" then (.kind + \"/\" + .path + \"/subscription/\" + .sub)
        else (.kind + \"/\" + .path + \"/subscription/\" + .sub + \"/skip_all\")
        end)} ] as $cand | {candidates: $cand, aged: ([ $cand[] |
        select(.first_seen <= $cutoff) ] | sort_by(.backlog) | reverse |
        .[0:$max]), in_grace: [ $cand[] | select(.first_seen > $cutoff) ],
        next_seen: ([ $cand[] | {key: .key, value: .first_seen} ] |
        from_entries)}') | first | toJson }}"

  - id: plan_counts
    type: io.kestra.plugin.core.output.OutputValues
    description: Pull the counts the gates branch on out of the plan.
    values:
      total_subscriptions: "{{ outputs.inventory.values.subscriptions | jq('length') | first }}"
      candidate_count: "{{ outputs.build_plan.values.plan | jq('.candidates | length')
        | first }}"
      aged_count: "{{ outputs.build_plan.values.plan | jq('.aged | length') | first }}"
      in_grace_count: "{{ outputs.build_plan.values.plan | jq('.in_grace | length') | first }}"
      aged_backlog_total: "{{ outputs.build_plan.values.plan | jq('[.aged[] |
        .backlog] | add // 0') | first }}"
      aged_summary: "{{ outputs.build_plan.values.plan | jq('[.aged[] | (.sub + \" on
        \" + .path + \" backlog=\" + (.backlog | tostring) + \" consumers=\" +
        (.consumers | tostring) + \" first_seen=\" + .first_seen)] | join(\";
        \")') | first }}"
      in_grace_summary: "{{ outputs.build_plan.values.plan | jq('[.in_grace[] | (.sub
        + \" on \" + .path + \" backlog=\" + (.backlog | tostring) + \"
        first_seen=\" + .first_seen)] | join(\"; \")') | first }}"
      aged_json: "{{ outputs.build_plan.values.plan | jq('.aged') | first | toJson }}"
      next_seen_json: "{{ outputs.build_plan.values.plan | jq('.next_seen') | first |
        toJson }}"

  - id: log_audit
    type: io.kestra.plugin.core.log.Log
    message: |
      Pulsar orphaned subscription audit for {{ inputs.tenant }}/{{ inputs.pulsar_namespace }}
      subscriptions inspected: {{ outputs.plan_counts.values.total_subscriptions }}
      consumerless at or above {{ inputs.min_backlog_messages }} messages: {{ outputs.plan_counts.values.candidate_count }}
      past the {{ inputs.grace_period_hours }}h grace window: {{ outputs.plan_counts.values.aged_count }}
      still inside the grace window: {{ outputs.plan_counts.values.in_grace_count }}
      messages pinned by the aged set: {{ outputs.plan_counts.values.aged_backlog_total }}
      aged targets: {{ outputs.plan_counts.values.aged_summary }}
      spared by the age gate: {{ outputs.plan_counts.values.in_grace_summary }}
      action: {{ inputs.remediation_action }} / dry_run={{ inputs.dry_run }} / auto_remediate={{ inputs.auto_remediate }}

  - id: save_first_seen
    type: io.kestra.plugin.core.kv.Set
    description: Persist the first-seen clock so a restarting consumer is never
      reaped on first sighting.
    key: "{{ render(vars.first_seen_key) }}"
    value: "{{ outputs.plan_counts.values.next_seen_json }}"

  - id: alert_findings
    type: io.kestra.plugin.core.flow.If
    condition: "{{ (outputs.plan_counts.values.candidate_count | number) > 0 }}"
    then:
      - id: notify_findings
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "Pulsar backlog held by subscriptions with no consumers in {{ inputs.tenant }}/{{ inputs.pulsar_namespace }}: {{ outputs.plan_counts.values.candidate_count }} found, {{ outputs.plan_counts.values.aged_count }} past the {{ inputs.grace_period_hours }}h grace window, {{ outputs.plan_counts.values.aged_backlog_total }} messages pinned. Targets: {{ outputs.plan_counts.values.aged_summary }}"
          }

  - id: decide
    type: io.kestra.plugin.core.flow.If
    condition: "{{ (outputs.plan_counts.values.aged_count | number) > 0 }}"
    then:
      - id: gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: hold
            type: io.kestra.plugin.core.flow.Pause
            description: >-
              Both actions are irreversible for the messages involved, so a
              human confirms the target list before any admin write.
            onResume:
              - id: approved
                type: BOOL
                displayName: Clear these subscriptions now

          - id: apply
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.hold.onResume.approved ?? false }}"
            then:
              - id: remediate_subscriptions
                type: io.kestra.plugin.core.flow.Loop
                description: One subscription at a time so two writes never race on the same
                  topic.
                concurrencyLimit: 1
                values: "{{ outputs.plan_counts.values.aged_json }}"
                tasks:
                  - id: live_stats
                    type: io.kestra.plugin.core.http.Request
                    description: >-
                      Read the topic's live stats. The namespace sweep uses the
                      broker's cached stats snapshot, which lags by the broker's
                      own statsUpdateFrequencyInSecs, so the state is confirmed
                      against a live read before anything is written.
                    method: GET
                    uri: "{{ inputs.admin_url }}/admin/v2/{{ fromJson(item.value).kind }}/{{
                      fromJson(item.value).path }}/stats"
                    headers:
                      Authorization: "{{ inputs.auth_mode == 'token' ? 'Bearer ' ~
                        secret('PULSAR_ADMIN_TOKEN') : 'Bearer none' }}"

                  - id: live_state
                    type: io.kestra.plugin.core.output.OutputValues
                    values:
                      state: "{{ {'body': outputs.live_stats.body, 'sub': fromJson(item.value).sub} |
                        toJson | jq(render(vars.subscription_program)) | first |
                        toJson }}"

                  - id: still_orphaned
                    type: io.kestra.plugin.core.flow.If
                    condition: "{{ (outputs.live_state.values.state | jq('.present') | first) and
                      (outputs.live_state.values.state | jq('.consumers') |
                      first) == 0 and (outputs.live_state.values.state |
                      jq('.backlog') | first) >= inputs.min_backlog_messages }}"
                    then:
                      - id: sample_doomed_messages
                        type: io.kestra.plugin.pulsar.Reader
                        description: >-
                          Read the oldest retained messages on the topic through
                          the broker, without a subscription and without acking,
                          so the operator keeps evidence of what the write
                          discards.
                        uri: "{{ inputs.broker_uri }}"
                        topic: "{{ fromJson(item.value).topic }}"
                        deserializer: STRING
                        maxRecords: "{{ inputs.sample_messages }}"
                        maxDuration: PT20S
                        pollDuration: PT2S

                      - id: log_sample
                        type: io.kestra.plugin.core.log.Log
                        message: >-
                          {{ fromJson(item.value).sub }} on {{
                          fromJson(item.value).path }} confirmed live as {{
                          outputs.live_state.values.state }}, first seen {{
                          fromJson(item.value).first_seen }}. Sampled {{
                          outputs.sample_doomed_messages.messagesCount }}
                          messages to {{ outputs.sample_doomed_messages.uri }}
                          before running {{ inputs.remediation_action }}.

                      - id: pick_action
                        type: io.kestra.plugin.core.flow.If
                        condition: "{{ inputs.remediation_action == 'skip_all' }}"
                        then:
                          - id: clear_backlog
                            type: io.kestra.plugin.core.http.Request
                            description: Acknowledge every pending message so the ledgers stop being pinned.
                            method: POST
                            allowFailure: true
                            uri: "{{ inputs.admin_url }}/admin/v2/{{ fromJson(item.value).action_path }}"
                            headers:
                              Authorization: "{{ inputs.auth_mode == 'token' ? 'Bearer ' ~
                                secret('PULSAR_ADMIN_TOKEN') : 'Bearer none' }}"
                        else:
                          - id: drop_subscription
                            type: io.kestra.plugin.core.http.Request
                            description: Delete the subscription so nothing references the ledgers any more.
                            method: DELETE
                            allowFailure: true
                            uri: "{{ inputs.admin_url }}/admin/v2/{{ fromJson(item.value).action_path }}"
                            headers:
                              Authorization: "{{ inputs.auth_mode == 'token' ? 'Bearer ' ~
                                secret('PULSAR_ADMIN_TOKEN') : 'Bearer none' }}"

                      - id: verify_stats
                        type: io.kestra.plugin.core.http.Request
                        description: Read the live topic stats again rather than trusting the write.
                        method: GET
                        uri: "{{ inputs.admin_url }}/admin/v2/{{ fromJson(item.value).kind }}/{{
                          fromJson(item.value).path }}/stats"
                        headers:
                          Authorization: "{{ inputs.auth_mode == 'token' ? 'Bearer ' ~
                            secret('PULSAR_ADMIN_TOKEN') : 'Bearer none' }}"

                      - id: verify_state
                        type: io.kestra.plugin.core.output.OutputValues
                        values:
                          state: "{{ {'body': outputs.verify_stats.body, 'sub': fromJson(item.value).sub}
                            | toJson | jq(render(vars.subscription_program)) |
                            first | toJson }}"

                      - id: log_verify
                        type: io.kestra.plugin.core.log.Log
                        message: >-
                          After {{ inputs.remediation_action }} on {{
                          fromJson(item.value).sub }}: before {{
                          outputs.live_state.values.state }}, after {{
                          outputs.verify_state.values.state }}. A present of
                          false means the subscription is gone from the broker.

                      - id: check_cleared
                        type: io.kestra.plugin.core.flow.If
                        condition: "{{ (outputs.verify_state.values.state | jq('.present') | first) and
                          (inputs.remediation_action == 'unsubscribe' or
                          (outputs.verify_state.values.state | jq('.backlog') |
                          first) > inputs.target_backlog_messages) }}"
                        then:
                          - id: not_cleared
                            type: io.kestra.plugin.core.execution.Fail
                            errorMessage: >-
                              {{ inputs.remediation_action }} did not land on {{
                              fromJson(item.value).sub }}. The live re-read
                              still shows {{ outputs.verify_state.values.state
                              }}.
                    else:
                      - id: consumer_returned
                        type: io.kestra.plugin.core.log.Log
                        level: WARN
                        message: >-
                          Skipping {{ fromJson(item.value).sub }} on {{
                          fromJson(item.value).path }}. The live re-read says {{
                          outputs.live_state.values.state }}, so it no longer
                          matches the orphan test the sweep flagged it on.
                          Nothing was written.
            else:
              - id: declined
                type: io.kestra.plugin.core.log.Log
                level: WARN
                message: >-
                  Approval was declined or left unanswered, so nothing was
                  cleared. Still pinned: {{
                  outputs.plan_counts.values.aged_summary }}
        else:
          - id: plan_only
            type: io.kestra.plugin.core.log.Log
            message: >-
              Read-only run. {{ inputs.remediation_action }} would run against
              {{ outputs.plan_counts.values.aged_count }} subscriptions holding
              {{ outputs.plan_counts.values.aged_backlog_total }} messages: {{
              outputs.plan_counts.values.aged_summary }}. Set auto_remediate
              true with dry_run false to arm the write.
    else:
      - id: nothing_to_clear
        type: io.kestra.plugin.core.log.Log
        message: >-
          No subscription in {{ inputs.tenant }}/{{ inputs.pulsar_namespace }}
          is both consumerless past {{ inputs.grace_period_hours }}h and at or
          above {{ inputs.min_backlog_messages }} messages. {{
          outputs.plan_counts.values.in_grace_count }} are inside the grace
          window.

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "pulsar-orphaned-subscription-backlog-clear-guard failed on {{ inputs.tenant }}/{{ inputs.pulsar_namespace }}. Execution {{ execution.id }}, task {{ task.id }}. Check whether the admin write landed before running it again."
      }

triggers:
  - id: nightly_backlog_audit
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 3 * * *"
    disabled: true
    inputs:
      dry_run: true
      auto_remediate: false

A Pulsar subscription with zero consumers does not stop costing anything. The broker keeps every unacknowledged message for it, the backlog grows without bound. The ledgers behind those messages can never be reclaimed either, because Pulsar will not delete a ledger that any subscription still references. A service decommissioned months ago can hold terabytes this way. The only sign is a storage graph that climbs while nothing reads. This blueprint finds those subscriptions through the Pulsar admin REST API, holds them behind an age window so a consumer that is merely restarting is never touched, then clears them behind a dry run and a human approval before reading the live topic stats again to prove the backlog actually went away.

This blueprint was created by zkasuran.

Safety

The two actions are not interchangeable, so remediation_action is an explicit input and it defaults to the less destructive one.

skip_all acknowledges every pending message for that subscription. The data is gone for that consumer group, the ledgers stop being pinned. The subscription survives with an empty backlog, so a consumer that reconnects later keeps its identity and receives new messages from that point on.

unsubscribe deletes the subscription outright. The ledgers stop being pinned in the same way, but the cursor goes with it, so a consumer that reconnects under the same subscription name starts fresh at the topic's current position instead of resuming where it left off.

Both are irreversible for the messages involved. Nothing in Pulsar can un-acknowledge a skipped message or restore a deleted cursor, which is why the write sits behind auto_remediate, dry_run and a Pause. Four separate brakes sit on it:

The age gate. Each consumerless subscription gets a first-seen timestamp in Kestra KV. Nothing is eligible until it has been consumerless since before grace_period_hours. A subscription seen for the first time in this run is recorded and left alone, so a rolling consumer restart never looks like an abandoned service.
The live re-check. The namespace sweep reads the broker's cached stats snapshot, which lags by the broker's own statsUpdateFrequencyInSecs, so every target is read again from the live per-topic stats endpoint immediately before the write. If a consumer reconnected in the meantime the target is skipped and nothing is written.
The evidence read. Before each write the flow reads a sample of the topic's oldest retained messages through the Pulsar plugin, with no subscription and no acks, so there is a record of what the write discarded.
The re-read. After the write the live stats are read again and the run fails if the backlog is still above target_backlog_messages. It also fails if the subscription still exists at all when the action was unsubscribe.

How it works

GET /admin/v2/persistent/{tenant}/{namespace} lists the topics the namespace holds right now.
One call to /admin/v2/broker-stats/topics returns every topic the broker owns with its subscriptions, each carrying msgBacklog and its connected consumer list.
A jq pass reduces that tree to one row per subscription on a topic that is still in the live list, dropping Pulsar's own system topics and system subscriptions, then keeps the rows with zero consumers and a backlog at or above min_backlog_messages.
The first-seen snapshot in Kestra KV splits those into aged targets and in-grace ones. Targets are ordered biggest backlog first and capped at max_subscriptions.
Findings go to Slack whether or not anything is eligible, so the detection is useful on its own.
With dry_run true or auto_remediate false the run stops at a log of exactly what it would have called. That is the default.
Armed, a Pause waits for a human. On approval a Loop with concurrencyLimit: 1 walks the targets one at a time: live re-check, sample through io.kestra.plugin.pulsar.Reader, then POST .../subscription/{sub}/skip_all or DELETE .../subscription/{sub}.
Each target is re-read from the live stats endpoint straight after its own write. The run fails if the write did not land.

What you get

Storage you can actually reclaim. Once no subscription references a ledger, Pulsar's own retention and ledger trimming can delete it, so the broker's disk goes down instead of sideways. You also get a decision you can defend: the target list, the backlog each subscription held, the consumer counts the broker reported, the grace window each one passed, the sampled messages. It reports the live numbers from the re-read after the write.

The sampled messages land in Kestra internal storage as an ion file at {{ outputs.sample_doomed_messages.uri }}, one row per message, with {{ outputs.sample_doomed_messages.messagesCount }} as the count. The per-target before and after states are logged as JSON, for example {"present":true,"backlog":3074,"bytes":174569,"consumers":0} ahead of the write and {"present":true,"backlog":0,"bytes":0,"consumers":0} after it.

The age gate is the part that makes this safe to schedule. A backlog alert cannot tell a decommissioned exporter from a consumer fleet rolling through a deploy. Two sightings separated by the grace window can.

Who it's for

Platform and streaming teams running Pulsar for more than one product team, where subscriptions outlive the services that created them. If broker storage keeps climbing while message rates are flat, this is usually why.

Why orchestrate this with Kestra

The job is a sequence of admin calls with a decision in the middle, which is what a flow is good at. Kestra supplies the parts a cron script would have to grow for itself: a KV store for the first-seen clock that survives restarts, a Pause that puts a named human in front of an irreversible call, secrets for the admin token, a re-read that can fail the run. It keeps a log of every number the decision used. The Pulsar plugin reads from the same broker in the same flow, so the evidence and the remediation share one credential set and one audit trail.

Tested end to end

Run against Apache Pulsar 3.3.1 in standalone mode on Kestra 2.0.5. One topic held 2000 messages with several subscriptions on it: orphaned ones with zero consumers and a backdated first-seen, one with a live consumer. One more was freshly created and consumerless inside the grace window.

Healthy namespace, every subscription consumed: 0 candidates with backlogs well above the threshold, no Slack alert, no write. That proves the detector keys on the consumer count rather than the backlog.
Degraded namespace: the sweep reported the orphans at msgBacklog 2000 with 0 consumers, left the subscription with a live consumer alone. It left the fresh orphan to the age gate.
dry_run: true: the plan named the target and the admin API showed the backlog unchanged.
auto_remediate: true with dry_run: false and approval: skip_all took the backlog from 2000 to 0 and the live re-read confirmed it. unsubscribe removed the subscription and the re-read confirmed it was gone.
Approval declined: nothing was called and the backlog stayed where it was.
A consumer that reconnected during the pause: the live re-check skipped the target and nothing was written.
Re-verify failure: with a producer still writing to the topic, skip_all could not hold the backlog at the target, the live re-read caught it and the run failed instead of reporting success.
Failure path: an unreachable admin URL routed to the errors block and posted to Slack.

Every before and after number was read from the Pulsar admin API directly, not from the Kestra log.

Prerequisites

A reachable Pulsar admin REST endpoint, an account with namespace-level admin rights on the tenant. The broker service URL is needed for the plugin read. The admin role needs skip_all and unsubscribe permission on the namespace.

Secrets

PULSAR_ADMIN_TOKEN: bearer token for the Pulsar admin REST API. Used while auth_mode is token.
SLACK_WEBHOOK_URL: incoming webhook for findings and failures.

Inputs

admin_url: base URL of the Pulsar admin REST API, for example http://pulsar-broker:8080.
broker_uri: broker service URL for the plugin read, for example pulsar://pulsar-broker:6650.
auth_mode: token to send the bearer token, none for a cluster with authentication disabled.
tenant, pulsar_namespace: the namespace to sweep.
min_backlog_messages: backlog at or above which a consumerless subscription is a candidate.
grace_period_hours: how long a subscription must stay consumerless before it can be touched.
target_backlog_messages: backlog the live re-read must show after a skip_all.
remediation_action: skip_all keeps the subscription with an empty backlog, unsubscribe deletes it.
max_subscriptions: cap on how many are acted on per run.
sample_messages: how many messages to read as evidence before each write.
dry_run, auto_remediate: the safety gates. Defaults are read-only.

Quick start

Add PULSAR_ADMIN_TOKEN and SLACK_WEBHOOK_URL to your namespace secrets.
Run it with the defaults against one namespace. Nothing is written, so read the log and see what it found.
Lower min_backlog_messages until the audit lines up with what you know about that namespace, then raise it to the level you want paged on.
Leave it on the disabled nightly schedule for a few days so the first-seen clock fills in and the grace window means something.
When a target looks right, run it once with auto_remediate: true and dry_run: false and approve the pause by hand.

How to extend

/admin/v2/broker-stats/topics reports the topics the broker you called owns. On a multi-broker cluster run the flow once per broker admin URL. The alternative is one run per namespace behind the proxy, looping the namespaces.
The live per-subscription stats also carry lastConsumedTimestamp and lastAckedTimestamp. Fold either into the eligibility test for a second idle signal next to the KV first-seen clock.
Swap the Slack task for PagerDuty or Opsgenie if the reclaim sits on an incident path.
Add a backlog quota on the namespace with POST /admin/v2/namespaces/{tenant}/{ns}/backlogQuota so new subscriptions cannot repeat the growth. Keep this flow for the ones that already did.
Point the pause at the team that owns the namespace by sending the target list to their channel first.

Clear Pulsar backlogs held by subscriptions nobody consumes

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?

Clear Pulsar backlogs held by subscriptions nobody consumes

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 Infrastructure Blueprints

Oracle Recycle Bin Space Reclaim Guard

Oracle Stale Optimizer Statistics Gather Guard

Oracle Sequence MAXVALUE Exhaustion Guard

Oracle Tablespace Autoextend Exhaustion Guard

Ory Kratos Identity Service Readiness & Health Sentinel

Send a notification via Opsgenie when a workflow fails

More Data Blueprints

AWS Glue Stuck Job Run Sentinel

Microsoft Azure Cosmos DB 429 Request Unit Rate Limit Sentinel

Glue ETL, Athena Validation, DynamoDB Cache

Monthly NYC Taxi Metrics with DuckDB as a Backfill Partition Job

Autoscale EMR Serverless Spark on a CloudWatch Backlog

Amazon DynamoDB Throttled Requests and Capacity Starvation Sentinel

New to Kestra?