Apache Pulsar transport

Last updated: October 25, 2026

com.babelqueue:babelqueue-pulsar is an Apache Pulsar transport on the Java 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

Maven:

<dependency>
    <groupId>com.babelqueue</groupId>
    <artifactId>babelqueue-pulsar</artifactId>
    <version>1.2.0</version> <!-- 1.2.0+: maxAttempts, DLQ, client(...) -->
</dependency>

Requirements: Java 17+. It pulls babelqueue-core and org.apache.pulsar:pulsar-client transitively. You supply the Pulsar Producer / Consumer.

Produce

import com.babelqueue.pulsar.PulsarPublisher;
import org.apache.pulsar.client.api.*;
import java.util.Map;

// publish(...) throws the checked PulsarClientException — declare or catch it.
PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:6650").build();
Producer<byte[]> producer = client.newProducer().topic("orders").create();

String id = PulsarPublisher.create(producer)
    .publish("urn:babel:orders:created", Map.of("order_id", 1042L));

publish(urn, data) returns the message meta.id; overloads add a traceId and a relative Duration delay (native deliverAfter).

Consume

import com.babelqueue.pulsar.PulsarConsumer;
import org.apache.pulsar.client.api.*;

Consumer<byte[]> sub = client.newConsumer()
    .topic("orders").subscriptionName("babelqueue")
    .subscriptionType(SubscriptionType.Shared).subscribe();

// Optional: a producer on the dead-letter topic turns dead-lettering on.
Producer<byte[]> dlq = client.newProducer().topic("orders.dlq").create();

PulsarConsumer consumer = PulsarConsumer.builder(sub)
    .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
    .handler("urn:babel:orders:created", (envelope, message) -> {
        // envelope.data(), envelope.traceId(), envelope.attempts() ...
    })
    .onError((error, envelope, message) -> error.printStackTrace())
    .build();

while (running) {
    consumer.poll(); // throws the checked PulsarClientException — declare or catch it
}
consumer.close();    // closes the retry producers the consumer created

Retry, max attempts and the DLQ

Redelivery is bounded by the SDK. Pulsar has no broker-side redelivery limit and this SDK configures no native DeadLetterPolicy (one you add yourself is a second, uncoordinated limit), 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 throws IllegalArgumentException.

• 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 the same topic with bq-attempts + 1 (the body’s attempts kept in sync, the other properties kept) and then acknowledges the original, but only after the re-publish succeeded; a failed write is reported to onError and the original is negatively acknowledged, so nothing is lost. A negative acknowledgement advances getRedeliveryCount() only on Shared and Key_Shared subscriptions, so the re-publish is what makes the cap hold on every subscription type, including the client default, Exclusive.
• Giving the consumer a producer. client(PulsarClient) makes the consumer create the retry producer itself, one per topic on the first retry. retryProducer(Producer<byte[]>) supplies your own, which wins over the client (you own it; close() leaves it open). Without either, a retry is a negative acknowledgement, which only terminates on Shared and Key_Shared. build() therefore throws IllegalStateException unless you pass a client, a retryProducer, or declare subscriptionType(SubscriptionType.Shared) (or Key_Shared) on the builder. An undeclared type counts as Exclusive, the fail-closed choice, because the Consumer interface does not expose its own subscription type.
• 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, original_queue, lang: "java"), keeping the bq-job route the delivery arrived on. Without it the message is acknowledged and dropped, and the drop is logged at WARNING. The original is acknowledged only after the DLQ write succeeded; a failed write is reported to onError and the message is negatively acknowledged, so the write is tried again.
• Poison. A body that is not valid UTF-8, not a JSON object, or that has no valid job / urn identity (or is otherwise not a conformant envelope) is poison. It never reaches a handler. With a deadLetterProducer it is written to <queue>.dlq at once, without retry, as a poison record (reason: "poison", the original body kept: an object body gets a dead_letter member, anything else is wrapped under raw, as base64 plus raw_encoding when it is not valid UTF-8). Without one 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 (1.2.0). Before 1.2.0 a throwing handler was negatively acknowledged 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. A consumer built on a Shared subscription must add .client(client) or .subscriptionType(SubscriptionType.Shared).

Behaviour change (ADR-0037, since 1.3.0). A message rejected by the bq-schema-version gate (below) used to be negatively acknowledged on every delivery, outside maxAttempts, so it could be redelivered without limit. It is now settled on its first delivery.

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(.., getRedeliveryCount()).

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, not version-gated.

A rejected message is settled on its first delivery, whatever bq-attempts, the redelivery count or maxAttempts say. No handler and no onUnknownUrn hook runs, and it is never retried or re-published. onError is called with a null envelope.

This consumer has no unknown-URN strategy option, so it behaves as fail. A reader of <queue>.dlq recognises the raw record by the absence of a dead_letter block.

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 getRedeliveryCount() is read as a cross-check: the handler sees max(<bq-attempts or body attempts>, getRedeliveryCount()). getRedeliveryCount() is 0-based, so it maps directly with no −1. The Pulsar Producer / Consumer / Message interfaces are mocked with Mockito 5 — no Pulsar, no network. The envelope is unchanged (schema_version stays 1); Apache Pulsar is purely additive.