Audit Pulsar Dead-Letter Topic Message Accumulation and Quarantine

In distributed streaming architectures powered by Apache Pulsar, consumer applications process continuous high-throughput event feeds. When unhandled payload formats, deserialization errors, or schema regressions occur, Pulsar's negative acknowledgment and dead-letter policy safeguards the main topic by redirecting failing records to a designated Dead-Letter Topic (DLQ). While this prevents pipeline head-of-line blocking, dead-letter messages often accumulate unnoticed in the background, resulting in silent business data loss, dropped transactional events, and unfulfilled customer requests.

This blueprint implements an automated Day 2 Operations sentinel flow that audits Apache Pulsar dead-letter topics on a recurring schedule. It polls the configured DLQ topic to detect quarantined poison-pill records, persists the failed payload samples into Kestra internal storage for forensic analysis, evaluates accumulation against warning thresholds, and dispatches real-time Slack alerts to platform teams before retention windows expire.

How it works

1. periodic_dlq_audit (io.kestra.plugin.core.trigger.Schedule) triggers the audit pass every 2 hours (shipped disabled for safe initial onboarding).
2. probe_dead_letter_topic (io.kestra.plugin.pulsar.Consume) connects to the Pulsar broker over binary protocol (pulsar://), binding to the specified dead-letter topic using an audit subscription to sample accumulated messages up to max_sample_records.
3. evaluate_dlq_status (io.kestra.plugin.core.flow.Switch) evaluates whether messagesCount breaches dlq_alert_threshold.
4. When within acceptable operating boundaries (HEALTHY), log_dlq_healthy records that 0 poison pills were found during the polling window.
5. When failed messages are detected (DLQ_BACKLOG_DETECTED), alert_dlq_backlog posts an urgent incident notification to Slack via io.kestra.plugin.slack.notifications.SlackIncomingWebhook with the message count and storage URI, and log_quarantine_telemetry logs remediation instructions.
6. log_audit_completion writes a final audit recap for execution tracking and audit lineage.
7. The flow-level errors block alerts Slack immediately if broker connectivity, DNS resolution, or authentication fails.

What you get

• Proactive visibility into quarantined streaming events before retention policies permanently drop data.
• Automatic persistence of failed message samples in Kestra storage for post-incident debugging.
• Real-time Slack notifications with exact topic names, record counts, and payload storage links.
• Configurable polling durations and batch sampling sizes that prevent broker resource starvation.
• Flow-level error alerting safeguarding audit pipeline reliability.

Who it's for

• Platform Engineers and Site Reliability Engineers operating Apache Pulsar message clusters.
• Streaming Data Engineers maintaining real-time event-driven microservices and ingestion pipelines.
• DevOps Teams monitoring messaging infrastructure for silent consumer processing failures.

Why orchestrate this with Kestra

Monitoring Pulsar dead-letter topics with ad-hoc cron jobs or detached Python scripts requires managing Pulsar client C++ library bindings, credential rotation, error handling, and notification integrations in fragile custom code. Kestra provides a declarative workflow engine with native Pulsar protocol integration, centralized secret management, conditional branching, execution history, and native Slack notifications without operating dedicated monitoring daemons.

Prerequisites

• A running Apache Pulsar cluster (v2.10+ or v3.x standalone/distributed deployment).
• An active Pulsar topic with a configured dead-letter policy (e.g. persistent://public/default/orders-dlq).
• A Slack Incoming Webhook URL configured in Kestra secrets.

Local Docker Container Setup

To run Apache Pulsar standalone locally for development and testing without host port conflicts:

docker run -d --name pulsar-dev -p 6650:6650 -p 8080:8080 \
  apachepulsar/pulsar:3.2.0 bin/pulsar standalone

If Kestra is running inside a Docker network, connect the Pulsar container to Kestra's network:

docker network ls
docker network connect YOUR_KESTRA_NETWORK pulsar-dev

Secrets

• PULSAR_AUTH_TOKEN: Optional JWT token for authenticated Pulsar clusters (not required for local standalone dev).
• SLACK_WEBHOOK: Incoming Webhook URL for Slack alerting.

Note for Kestra Open-Source (OSS): Secrets are configured as base64-encoded environment variables formatted as SECRET_<NAME> (e.g. SECRET_SLACK_WEBHOOK).

Inputs

Name	Type	Default	Allowed Values	Description
pulsar_service_url	STRING	"pulsar://localhost:6650"	Valid pulsar:// URI	Pulsar broker binary endpoint
dead_letter_topic	STRING	"persistent://public/default/orders-dlq"	Valid topic string	Target dead-letter topic name
subscription_name	STRING	"kestra-dlq-audit-sub"	String identifier	Audit consumer subscription name
poll_timeout	STRING	"PT3S"	ISO-8601 duration (PT2S, PT5S)	Maximum poll wait duration
max_sample_records	INT	50	Positive integer	Maximum batch size of sampled records
dlq_alert_threshold	INT	0	Non-negative integer	Alert threshold for accumulated messages

Outputs

• {{ outputs.probe_dead_letter_topic.messagesCount }}: Total number of dead-letter records consumed during the audit poll.
• {{ outputs.probe_dead_letter_topic.uri }}: Internal Kestra storage URI containing the raw JSON/ION payload of the consumed dead-letter records.

Quick start

1. Configure SLACK_WEBHOOK in your Kestra namespace secrets.
2. Import this blueprint into your Kestra instance.
3. Run a manual execution with dead_letter_topic pointing to your Pulsar DLQ topic to verify connectivity and message sampling.
4. Enable the periodic_dlq_audit schedule trigger to run automated recurring audits.

How to extend

• Add automated payload schema validation by passing outputs.probe_dead_letter_topic.uri into a JSON Schema validator task.
• Chain an automated replay pipeline using io.kestra.plugin.pulsar.Produce to re-publish quarantined messages back to the primary topic once downstream consumer bugs are patched.
• Integrate with PagerDuty or Opsgenie in the DLQ_BACKLOG_DETECTED branch for critical payment or security telemetry queues.

Pitfalls

• Subscription Type and Multiple Workers: The io.kestra.plugin.pulsar.Consume task defaults to an Exclusive subscription. Ensure concurrent executions or other services do not use the same subscription_name, or switch to a Shared subscription if concurrent workers are required.
• Message Acknowledgment: The Consume task acknowledges messages after reading. If you wish to inspect messages without acknowledging or removing them from the DLQ topic, configure a dedicated audit subscription or use io.kestra.plugin.pulsar.Reader instead.
• Pulsar Service vs HTTP Admin Port: Apache Pulsar exposes port 6650 for binary client connections (pulsar://) and port 8080 for HTTP admin REST API (http://). Ensure pulsar_service_url uses port 6650.

Links

• Kestra Apache Pulsar Plugin Documentation
• Apache Pulsar Dead Letter Topic Guide
• Apache Pulsar Client Configuration
• Kestra Slack Notification Plugin