Apache Pulsar multi-tenant backlog quarantine and priority event router

A single shared Pulsar topic serving many tenants has no built-in way to treat an enterprise customer's traffic differently from a free-tier customer's. This blueprint drains the shared topic on a schedule, reads the tenant_tier field carried in each message's JSON payload, and branches: ENTERPRISE traffic is logged as priority-handled, everything else is republished to a quarantine topic instead of being processed inline or silently dropped. One Slack message reports the outcome of every run.

How it works

1. hourly_backlog_drain (io.kestra.plugin.core.trigger.Schedule) runs once an hour. Shipped disabled so you can validate a manual run first.
2. consume_tenant_events (io.kestra.plugin.pulsar.Consume) batch-receives up to maxRecords: 50 JSON messages from tenant_topic under subscription kestra-backlog-router. Its own output is only messagesCount and a storage uri to an Ion file, one record per message, each shaped {key, value, properties, topic, eventTime, messageId}, not a messages list.
3. extract_tenant_json (io.kestra.plugin.serdes.json.IonToJson) converts that Ion file to a JSON array so its contents can be queried with jq.
4. route_by_tenant_tier (io.kestra.plugin.core.flow.Switch) reads {{ read(outputs.extract_tenant_json.uri) | jq('.[0].value.tenant_tier') | first | default('FREE_TIER') }}, the first consumed message's tenant_tier field, and branches:
   • ENTERPRISE: process_enterprise_event logs priority handling.
   • FREE_TIER: quarantine_free_tier (io.kestra.plugin.pulsar.Produce) republishes the same consumed records to quarantine_topic.
   • default: log_unclassified_tenant logs that the resolved value was neither ENTERPRISE nor FREE_TIER (an unrecognized tier string; see Pitfalls for what an empty batch resolves to instead).
5. notify_slack posts one summary per run: messages consumed and the tier that was routed.

What you get

• A scheduled drain of a shared multi-tenant Pulsar topic with explicit priority routing instead of first-come-first-served processing.
• A quarantine lane for non-priority traffic that keeps messages in Pulsar (on quarantine_topic) rather than discarding them.
• A default branch that surfaces batches with no recognizable tier instead of silently falling through.
• One Slack message per run with the consumed count and the tier that was acted on.

Who it's for

• Platform teams running shared Pulsar infrastructure across customers with different SLA tiers.
• SaaS teams who need enterprise traffic visibly prioritized without standing up separate topics per tenant.
• Anyone who has had a free-tier burst starve enterprise traffic on a shared topic and wants a routing seam to fix it.

Why orchestrate this with Kestra

Pulsar's own subscription and topic model has no concept of "read this field and branch my pipeline on it", that decision, and what happens in each branch, has to live outside the broker. Kestra's Switch makes that branch a first-class, inspectable step; the schedule and execution history mean every drain (priority-handled, quarantined, or unclassified) is reviewable after the fact, and the same flow can fan out to more than two branches or add new tiers by adding cases entries.

Prerequisites

• A reachable Apache Pulsar cluster. Locally, standalone mode works: docker run -p 6650:6650 -p 8081:8080 apachepulsar/pulsar:latest bin/pulsar standalone, the web admin port is remapped to host port 8081 (-p 8081:8080) specifically to avoid colliding with Kestra's own UI on localhost:8080; the client protocol port 6650 is unaffected and is what pulsar_uri connects to.
• tenant_topic and quarantine_topic must already exist (or auto-topic-creation enabled on the namespace) and carry messages whose JSON body includes a tenant_tier field, e.g. {"tenant_tier": "ENTERPRISE", ...}.
• A Slack incoming webhook for the summary notification.

Secrets

• SLACK_WEBHOOK_URL: Slack incoming webhook used by notify_slack.
• 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/...' | 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.

Inputs

• pulsar_uri (STRING, default pulsar://localhost:6650): Pulsar broker connection string.
• tenant_topic (STRING, default persistent://public/default/tenant-inbound): multi-tenant topic to drain.
• quarantine_topic (STRING, default persistent://public/default/tenant-quarantine): destination for non-priority batches.

Outputs

• outputs.consume_tenant_events.messagesCount / .uri: how many messages were pulled this run, and the raw Ion file of consumed records.
• outputs.extract_tenant_json.uri: the same records as a JSON file, used for the routing decision.
• outputs.quarantine_free_tier.messagesCount: set only when the FREE_TIER branch ran, the number of records republished to quarantine_topic.

Quick start

1. Start Pulsar locally with the command above, or point at an existing cluster.
2. Add the SLACK_WEBHOOK_URL secret.
3. Publish a few JSON messages with a tenant_tier field onto tenant_topic.
4. Run the flow manually and confirm the Slack message and branch taken match what you published.
5. Enable hourly_backlog_drain once a manual run looks correct.

How to extend

• Replace the first-message-decides-the-batch routing with a per-message loop (io.kestra.plugin.core.flow.ForEach over the JSON array) if different messages in the same batch carry different tiers.
• Add more cases (e.g. PRO_TIER) as your tenant model grows past two tiers.
• Feed quarantine_free_tier's output into a billing or throttling system instead of only Pulsar.
• Swap the Schedule trigger for io.kestra.plugin.pulsar.RealtimeTrigger for per-message routing instead of hourly batches, trading the batch efficiency of Consume for per-event latency.

Pitfalls

• The whole batch is routed by one message's tier. route_by_tenant_tier's value reads only .[0], the first record in the batch. A batch mixing ENTERPRISE and FREE_TIER messages is routed entirely by whichever happened to be first. This mirrors a deliberate simplification in this blueprint's routing key, not a bug in the Switch task; see "How to extend" for a per-message alternative.
• Consume's output has no messages field. It only exposes messagesCount (an Integer) and uri (an Ion file). Any expression written against outputs.consume_tenant_events.messages will fail to render, always go through extract_tenant_json's JSON file first.
• An empty batch's routing is UNVERIFIED, not something this blueprint confirmed by running it. On an empty batch, jq('.[0].value.tenant_tier') evaluates .[0] of an empty array to null, so first yields null; whether Kestra's default('FREE_TIER') filter treats that rendered null as "absent" (substituting FREE_TIER) or passes it through (landing in defaults instead) was not tested against a running Kestra instance. Either outcome still avoids an error, but if an empty poll reaching quarantine_free_tier and calling Produce against zero source records is not acceptable, branch on outputs.consume_tenant_events.messagesCount == 0 before the Switch instead of relying on this fallback.
• subscriptionName is shared and durable. Every run uses the same kestra-backlog-router subscription, so Pulsar tracks one cursor across runs, this is what makes the hourly drain incremental rather than replaying the same backlog. Changing the subscription name resets that cursor.

Links

• Apache Pulsar plugin docs
• Pulsar topic naming (persistent://tenant/namespace/topic)
• Kestra secrets (OSS vs Enterprise)
• Sibling blueprint: nats-event-validation-quarantine
• Sibling blueprint: kafka-dlq-quarantine-and-replay