Apache Pulsar Topic Activity Sentinel for Kestra

id: pulsar-topic-activity-sentinel
namespace: company.team
description: |
  On a schedule, non-destructively sample a Pulsar topic with a non-durable
  Reader (no subscription created, nothing acknowledged, no consumer group
  backlog touched) and classify the topic as SILENT, BURST, HEALTHY, or
  UNREACHABLE. Log clean runs and page Slack the moment a producer outage,
  a message flood, or a broker connectivity failure is detected.

inputs:
  - id: topic
    type: STRING
    defaults: "persistent://public/default/orders"
    displayName: "Pulsar Topic"
    description: "Fully qualified Pulsar topic to sample, e.g.
      persistent://tenant/namespace/topic."
  - id: pulsar_uri
    type: STRING
    defaults: "pulsar://localhost:6650"
    displayName: "Pulsar Service URI"
    description: "Pulsar broker service URL, e.g. pulsar://host:6650 or
      pulsar+ssl://host:6651."
  - id: lookback_window
    type: STRING
    defaults: "PT10M"
    displayName: "Lookback Window"
    description: "ISO-8601 duration passed to the reader's since property; how far
      back to look for existing messages on the topic."
  - id: probe_duration
    type: STRING
    defaults: "PT15S"
    displayName: "Probe Duration"
    description: "ISO-8601 duration; maximum wall-clock time the reader is allowed
      to run before it stops and returns its count."
  - id: min_expected_messages
    type: INT
    defaults: 1
    displayName: "Minimum Expected Messages"
    description: "Below this sampled count within the lookback window, the topic is
      classified SILENT (likely producer outage)."
  - id: max_expected_messages
    type: INT
    defaults: 5000
    displayName: "Maximum Expected Messages (Burst Ceiling)"
    description: "At or above this sampled count, the topic is classified BURST
      (possible retry storm or runaway producer). Also used as the reader's
      maxRecords cap."

triggers:
  - id: topic_activity_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: Sample the topic every 15 minutes. Shipped disabled; enable it once
      topic/pulsar_uri point at a real broker and a manual run has been
      verified.
    cron: "*/15 * * * *"
    disabled: true
    inputs:
      lookback_window: "PT10M"
      probe_duration: "PT15S"

tasks:
  - id: probe_topic_activity
    type: io.kestra.plugin.pulsar.Reader
    description: Use a non-durable reader to sample the topic without creating a
      subscription, so this check never consumes, acknowledges, or shows up in
      any real consumer group's backlog.
    uri: "{{ inputs.pulsar_uri }}"
    topic: "{{ inputs.topic }}"
    deserializer: STRING
    since: "{{ inputs.lookback_window }}"
    maxRecords: "{{ inputs.max_expected_messages }}"
    maxDuration: "{{ inputs.probe_duration }}"
    allowFailure: true

  - id: classify_topic_health
    type: io.kestra.plugin.core.debug.Return
    description: Classify the sampled count against the configured thresholds,
      treating a missing output as a broker or connectivity failure rather than
      silence.
    format: "{{ outputs.probe_topic_activity.messagesCount is not defined ?
      'UNREACHABLE' : (outputs.probe_topic_activity.messagesCount <
      inputs.min_expected_messages ? 'SILENT' :
      (outputs.probe_topic_activity.messagesCount >=
      inputs.max_expected_messages ? 'BURST' : 'HEALTHY')) }}"

  - id: route_by_health
    type: io.kestra.plugin.core.flow.Switch
    description: Route to a quiet log on a healthy sample or a Slack page for every
      other classification.
    value: "{{ outputs.classify_topic_health.value }}"
    cases:
      "HEALTHY":
        - id: log_healthy
          type: io.kestra.plugin.core.log.Log
          message: "Pulsar topic {{ inputs.topic }} is healthy: sampled {{
            outputs.probe_topic_activity.messagesCount }} message(s) in the last
            {{ inputs.lookback_window }}."
      "SILENT":
        - id: alert_silent_topic
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Page Slack when the topic produced fewer messages than expected, a
            likely producer outage.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "text": "Pulsar sentinel SILENT: topic {{ inputs.topic }} on {{ inputs.pulsar_uri }} returned only {{ outputs.probe_topic_activity.messagesCount }} message(s) in the last {{ inputs.lookback_window }} (minimum expected {{ inputs.min_expected_messages }}). The producer, the network, or the broker may be down. Execution {{ execution.id }}."
            }
      "BURST":
        - id: alert_burst_topic
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Page Slack when the sampled count hit the burst ceiling, a possible
            retry storm or runaway producer.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "text": "Pulsar sentinel BURST: topic {{ inputs.topic }} on {{ inputs.pulsar_uri }} hit {{ outputs.probe_topic_activity.messagesCount }} message(s) in the last {{ inputs.lookback_window }}, at or above the configured ceiling of {{ inputs.max_expected_messages }}. Check for a retry storm or runaway producer. Execution {{ execution.id }}."
            }
      "UNREACHABLE":
        - id: alert_unreachable_topic
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Page Slack when the reader task could not return a message count at
            all, indicating the broker or topic is unreachable.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "text": "Pulsar sentinel UNREACHABLE: the reader on topic {{ inputs.topic }} at {{ inputs.pulsar_uri }} did not return a message count (connection failure, auth failure, or missing topic). Execution {{ execution.id }}."
            }
    defaults:
      - id: log_unclassified
        type: io.kestra.plugin.core.log.Log
        message: "Pulsar sentinel completed with an unexpected classification: {{
          outputs.classify_topic_health.value }}."

  - id: log_audit_result
    type: io.kestra.plugin.core.log.Log
    description: Record a single audit line for every run regardless of branch
      taken, so the check's own history stays queryable independent of Slack.
    message: "Pulsar sentinel audit: topic={{ inputs.topic }} status={{
      outputs.classify_topic_health.value }} sampled_count={{
      outputs.probe_topic_activity.messagesCount is defined ?
      outputs.probe_topic_activity.messagesCount : 'n/a' }} execution={{
      execution.id }}."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert Slack when the sentinel flow itself fails outright, so a
      broken check is never mistaken for a healthy topic.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "Pulsar sentinel flow {{ flow.id }} FAILED (execution {{ execution.id }}). Topic activity on {{ inputs.topic }} is now unknown; check the task logs and the broker manually."
      }

outputs:
  - id: status
    type: STRING
    value: "{{ outputs.classify_topic_health.value }}"

An Apache Pulsar topic can stop receiving messages, or start flooding with them, and nothing in the broker itself pages anyone about either condition. This blueprint turns a scheduled, non-destructive sample into a Day 2 health gate: a non-durable io.kestra.plugin.pulsar.Reader looks back over a configurable window, classifies the sampled message count against two thresholds, and routes the result to a quiet log or a Slack page depending on what it finds.

How it works

probe_topic_activity (io.kestra.plugin.pulsar.Reader) opens a non-durable reader against topic at pulsar_uri. It looks back lookback_window (its since property), stops after max_expected_messages records or probe_duration, whichever comes first, and returns messagesCount. A non-durable reader creates no subscription, so this check never acknowledges a message, never advances a consumer group's cursor, and never shows up as backlog on a real subscription. allowFailure: true keeps a connection failure from aborting the flow outright.
classify_topic_health (io.kestra.plugin.core.debug.Return) evaluates a single Pebble expression against messagesCount: missing entirely becomes UNREACHABLE, below min_expected_messages becomes SILENT, at or above max_expected_messages becomes BURST, otherwise HEALTHY.
route_by_health (io.kestra.plugin.core.flow.Switch) branches on that classification: HEALTHY logs a quiet confirmation; SILENT, BURST, and UNREACHABLE each page Slack with the sampled count, the configured thresholds, and the execution id; an unreachable defaults fallback logs the raw value for any classification outside the four known cases.
log_audit_result (io.kestra.plugin.core.log.Log) writes one audit line per run, independent of which Switch branch fired, so the check's own history stays queryable even if Slack delivery fails.
The flow-level errors block pages Slack separately if the flow fails outright, so a broken sentinel is never mistaken for a quiet topic.
The topic_activity_schedule trigger samples every 15 minutes and ships disabled: true, with explicit inputs: overrides for lookback_window and probe_duration so the scheduled run's probe window is pinned regardless of this flow's own defaults.

What you get

A scheduled, non-destructive activity check for any Pulsar topic with zero custom consumer code.
A single status output (HEALTHY / SILENT / BURST / UNREACHABLE) other flows can gate on.
Three distinct Slack pages, one per failure mode, each carrying the sampled count and the threshold that tripped it.
An audit log line on every run, separate from Slack, so the check's own history survives a webhook outage.

Who it's for

Platform and data engineering teams running Apache Pulsar who have no orchestrator-side activity check today.
On-call engineers who want a producer-outage or retry-storm signal before it becomes a backlog incident.
Teams standardizing Day 2 monitoring across mixed messaging systems inside Kestra without writing a bespoke Pulsar admin client.

Why orchestrate this with Kestra

A cron job that opens a Pulsar client and exits is invisible the moment it succeeds, and a naive durable consumer used purely for monitoring would itself distort the backlog it is trying to observe. Kestra's non-durable Reader task avoids that trap, and wrapping it in a flow keeps every sample in the execution history: the sampled count becomes a typed output, the classification is a Switch branch you can extend with real remediation, and the whole check lives in version-controlled YAML instead of a standalone script nobody reviews.

Prerequisites

A running Apache Pulsar broker reachable at pulsar_uri, with the target topic already created.
A Slack incoming webhook for the SILENT, BURST, UNREACHABLE, and flow-failure alerts.

Secrets

SLACK_WEBHOOK_URL: Slack incoming webhook used for every alert branch and the flow-failure alert.

In Kestra OSS (no Enterprise secrets backend), secrets are supplied as environment variables prefixed SECRET_, base64-encoded, and read back in flows with {{ secret('NAME') }}, for example SECRET_SLACK_WEBHOOK_URL=$(echo -n 'https://hooks.slack.com/services/<placeholder>' | base64). This keeps the webhook out of the flow YAML but, per Kestra's own documentation, offers no encryption at rest or access control beyond the host environment; use the Enterprise secrets backend for stronger guarantees.

Quick start

Start a local Pulsar standalone broker, mapping its web port off host 8080 since Kestra's own UI already uses it: docker run -d --name pulsar-standalone -p 6650:6650 -p 9080:8080 apachepulsar/pulsar:3.2.2 bin/pulsar standalone
If Kestra is also running in Docker, attach the broker to Kestra's network so the worker can reach it by container name: docker network connect YOUR_KESTRA_NETWORK pulsar-standalone Then set pulsar_uri to pulsar://pulsar-standalone:6650. From a host-based Kestra install, use pulsar://localhost:6650 instead.
Create the topic and publish a few test messages: docker exec pulsar-standalone bin/pulsar-admin topics create persistent://public/default/orders docker exec pulsar-standalone bin/pulsar-client produce persistent://public/default/orders -m "test" -n 3
Add the SLACK_WEBHOOK_URL secret, set topic and pulsar_uri, and run the flow once manually to confirm status reads HEALTHY.
Enable topic_activity_schedule.

How to extend

Add an authenticationToken secret to probe_topic_activity if the broker requires token-based authentication.
Add a WARNING tier between SILENT and HEALTHY once you have real traffic baselines to tune against.
Push status into a KV store and alert only on state transitions instead of every run.
Point a second copy of this flow at a dead-letter or retry topic to watch it separately from the primary topic.

Inputs

topic: fully qualified Pulsar topic to sample.
pulsar_uri: Pulsar broker service URL.
lookback_window: ISO-8601 duration for how far back the reader looks (its since property).
probe_duration: ISO-8601 duration cap on how long the reader runs.
min_expected_messages: floor below which the topic is classified SILENT.
max_expected_messages: ceiling at or above which the topic is classified BURST, and the reader's maxRecords cap.

Outputs

status: one of HEALTHY, SILENT, BURST, UNREACHABLE.

Pitfalls

A non-durable reader samples raw message arrival; it does not read Pulsar's own backlog or consumer-lag metrics for any real subscription, so it cannot tell you how far behind an actual consumer group is, only whether the topic itself is producing traffic.
Setting max_expected_messages too low relative to real traffic produces false BURST alerts; tune it against an observed baseline before enabling the schedule.
since depends on the broker's recorded publish timestamps; a broker with significant clock skew against its producers can skew the lookback window.
This flow never sets subscriptionName and will not appear in, or interfere with, any production consumer group's backlog.