Apache Pulsar transport

Last updated: October 25, 2026

BabelQueue.Pulsar is an Apache Pulsar transport on the .NET core. It sends the canonical envelope as the message payload with the §5 property projection, and consumes by routing each message to a handler by URN — so a message it produces is consumed by any other BabelQueue SDK, and vice-versa.

Install

dotnet add package BabelQueue.Pulsar

Requirements: .NET 8. It pulls BabelQueue.Core and DotPulsar (a pure-C# Pulsar client) transitively.

Produce

using DotPulsar;
using DotPulsar.Extensions;
using BabelQueue.Pulsar;

await using var client = PulsarClient.Builder()
    .ServiceUrl(new Uri("pulsar://localhost:6650")).Build();

await using var producer = client.NewProducer(Schema.ByteArray).Topic("orders").Create();
var id = await new PulsarPublisher(producer)
    .PublishAsync("urn:babel:orders:created", new Dictionary<string, object?> { ["order_id"] = 1042 });

PublishAsync returns the message meta.id; pass a traceId to continue a trace, or a delay (TimeSpan) to schedule native delayed delivery (DeliverAtTime).

Consume

await using var consumer = client.NewConsumer(Schema.ByteArray)
    .Topic("orders").SubscriptionName("babelqueue")
    .SubscriptionType(SubscriptionType.Shared).Create();

// Optional: a producer on the dead-letter topic turns dead-lettering on.
await using var dlq = client.NewProducer(Schema.ByteArray).Topic("orders.dlq").Create();

var handlers = new Dictionary<string, BabelHandler>
{
    ["urn:babel:orders:created"] = (envelope, message, ct) =>
    {
        // envelope.Data, envelope.TraceId, envelope.Attempts ...
        return Task.CompletedTask;
    },
};
await using var babel = new PulsarConsumer(consumer, handlers, new PulsarConsumerOptions
{
    Client = client,            // the consumer builds its own retry producer from it
    MaxAttempts = 3,            // the default: 3 deliveries in total, the first included
    DeadLetterProducer = dlq,   // omit it and an exhausted message is dropped and logged
    OnError = (err, env, msg) => Console.Error.WriteLine(err),
});
await babel.RunAsync(cancellationToken);

Retry, max attempts and the DLQ

Redelivery is bounded by the SDK. Pulsar has no broker-side redelivery limit, and DotPulsar 5.3.1 has no client-side DeadLetterPolicy, so the consumer counts the attempts itself. MaxAttempts is the total number of deliveries a message gets, the first included; the default is 3 and 1 means no retry. A value below 1 makes the constructor throw ArgumentOutOfRangeException.

• Counted attempt. On every delivery the consumer takes the reconciled attempts (below) and adds one for the current delivery. A throwing handler, or an unmapped URN without an OnUnknownUrn hook, is retried while that counted attempt is below MaxAttempts.
• A retry is a re-publish. The consumer re-publishes the message to its own topic with bq-attempts + 1 (the body re-encoded by the codec with attempts in sync) and then acknowledges the original, but only after the publish succeeded; a failed publish leaves the original for redelivery, so nothing is lost. A negative acknowledgement advances RedeliveryCount only on Shared and Key_Shared subscriptions, so the re-publish is what makes the cap hold on every subscription type. A republished message loses its original order and any DeliverAt schedule.
• Giving the consumer a producer. Client (the IPulsarClient the consumer was built from) lets the consumer create the retry producer itself on consumer.Topic, which must be a single topic; PulsarConsumer implements IAsyncDisposable to dispose it. RetryProducer supplies your own instead (used as given, never disposed by the consumer). With neither, a retry is a negative acknowledgement, which only terminates on Shared and Key_Shared: the constructor throws InvalidOperationException for an Exclusive or Failover subscription without one of them, and also when the subscription type is not reported (it is then treated as Exclusive, the fail-closed choice), instead of looping at run time.
• At the cap. With DeadLetterProducer set (a producer on <queue>.dlq) the message is written there as the canonical envelope plus the dead_letter block (reason failed or unknown_urn, attempts = the counted value, lang: "dotnet"), keeping the bq-job route and the other incoming properties. Without it the message is acknowledged and dropped, and a trace warning is written. The original is acknowledged only after the DLQ write succeeded; a failed write is reported to OnError and the message stays unacknowledged, so it is redelivered.
• Poison. A body that never reaches a handler (not valid UTF-8, not a JSON object, no valid job / urn identity, any other non-conformant envelope, or a body with an unsupported meta.schema_version) is poison. A message rejected by the bq-schema-version property gate below is not poison: it is settled on its first delivery. With DeadLetterProducer it is written to <queue>.dlq at once, without retry, as a poison record (reason: "poison", the original bytes kept, the message’s properties carried over); a body that does not decode no longer gets rejected and redelivered forever. Without a producer it takes the bounded retry and is then dropped.
• OnUnknownUrn is unchanged. When the hook is set it runs and the message is acknowledged; the bounded path above is for an unmapped URN without a hook.

Behaviour change. Before this release a throwing handler was redelivered without a limit. With the defaults, a message that keeps failing is now dropped (or dead-lettered) at its third delivery; raise MaxAttempts to keep retrying longer.

Per §5.7, the consumer routes on the bq-job property (the URN), falling back to the body’s job / urn only when the property is absent or blank (externally produced messages). The body is still decoded and the handler receives the body’s envelope. If bq-job and the body’s job differ, the property still routes and the optional OnRoutingMismatch hook (since 1.1.0) is called with both URNs (a throwing hook is swallowed; settling is unaffected). attempts comes from bq-attempts when it is a non-negative integer (^[0-9]+$ that fits an int), otherwise from the body, and then max(.., RedeliveryCount).

Since 1.1.0, before decoding, the consumer version-gates on the bq-schema-version property (as the Pulsar binding requires; 1.0.0 decodes the body first and only rejects through the body check). A missing or blank property (the empty string, or only ASCII whitespace: space, tab, LF, VT, FF, CR) changes nothing: the body is decoded as usual. Any other value is compared with "1" exactly as sent, with no trimming, so "2", "x", "01" and " 1" are all rejected without decoding the body. A "1" property does not skip the body check: the decoded envelope still goes through the schema_version validation, and a body that fails it is poison.

A rejected message is settled on its first delivery (ADR-0037, since 1.3.0), whatever bq-attempts, RedeliveryCount or MaxAttempts say. OnError is notified with an empty envelope; no handler and no OnUnknownUrn hook runs, and it is never retried or re-published.

Behaviour change (ADR-0037, since 1.3.0). Until now a version-gated message took the bounded poison route: a bounded retry while no DeadLetterProducer was set, the poison record while one was, spending up to MaxAttempts deliveries on a message that can never succeed. It no longer does. This consumer has no unknown-URN strategy option, so it behaves as fail.

OnError and OnRoutingMismatch are synchronous Action delegates (OnUnknownUrn returns a Task and may be async). Do not assign an async lambda to them: it compiles to async void, so an exception thrown after its first await escapes the consumer’s try/catch and can crash the process. Keep hooks synchronous, or hand async work off yourself and observe its failures.

Contract mapping (§5)

Pulsar properties are string→string, so integers are stringified. Per the contract, bq-job carries the URN so a consumer can route on a property rather than the body’s job, and bq-attempts is the authoritative retry counter; the native RedeliveryCount is read as a cross-check: the handler sees max(<bq-attempts or body attempts>, RedeliveryCount). RedeliveryCount is 0-based, so it maps directly with no −1. The DotPulsar IProducer / IConsumer / IMessage interfaces are mockable, so the unit tests use Moq — no Pulsar, no network. The envelope is unchanged (schema_version stays 1); Apache Pulsar is purely additive.