Skip to main content

Subscription types and ordering

This guide chooses how the consumers of a Pulsar subscription share it, keeps the order of each key, and places keyed messages on partitions.

The example is a shipping service. It packs orders, tracks shipments, books payments and invoices orders, and each of these reads under a different subscription type.

The four types​

Every consumer of a subscription is a handler in one of the service's instances. The subscription type decides how those consumers share the messages:

TypeWho gets a messageOrder
Shared (default)any consumer, in turnnone
KeySharedthe consumer that holds the message's keyper key
Exclusivethe one consumer; a second one waits for itthe topic's order
Failoverthe active consumer; the others stand bythe topic's order

subscription_type on the descriptor sets the type. Every consumer of one subscription names the same type.

Shared: any consumer​

src/main.rs
/// One step in the life of a shipment.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "shipments")]
struct Shipment {
    order: u64,
    status: String,
}

#[derive(OutSlot)]
#[publishes(Shipment)]
struct Shipments;

/// Packs an order. Every instance of the service shares the subscription `packing`.
#[subscriber(PulsarSubscription::new("orders", "packing"))]
async fn pack(
    order: &Order,
    Out(shipments): Out<impl Publisher<Options = PulsarPublishOptions>, Shipments>,
) -> HandlerOutcome {
    let packed = Shipment {
        order: order.id,
        status: "packed".to_owned(),
    };
    // The order's id keys the event, so every event of one order keeps its place in line.
    let sent = shipments
        .message(&packed)
        .partition_key(order.id.to_string())
        .publish()
        .await;
    if sent.is_err() {
        return HandlerOutcome::retry();
    }
    tracing::info!(order.id, "order packed");
    HandlerOutcome::ack()
}

Shared hands each order to one consumer of packing, whichever is next. More instances of the service pack more orders at once, and two orders may finish in any order. That suits work whose messages do not depend on each other.

pack publishes a shipment event for each order. The partition_key step keys the event by the order's id. The step comes from the broker's prelude, so the slot is bounded Publisher<Options = PulsarPublishOptions> to offer it. A publish returns once the broker has stored the message.

KeyShared: in order per key​

src/main.rs
/// Follows each shipment. Events of one order arrive in the order they were published.
#[subscriber(
    PulsarSubscription::new("shipments", "tracking")
        .subscription_type(SubscriptionType::KeyShared),
    workers(8, by_key)
)]
async fn track(event: &Shipment) -> HandlerOutcome {
    tracing::info!(event.order, event.status, "shipment tracked");
    HandlerOutcome::ack()
}

KeyShared gives all the messages of one key to one consumer, so the events of one order reach track in the order pack published them. Different keys spread over the consumers, so the instances share the work. The server assigns the keys and moves some of them when a consumer joins or leaves.

Inside one instance, workers(8, by_key) runs eight lanes, and the deliveries of one key run one after another on one lane. The lane key is the message's partition key, so the order of each key holds within the process too. In the manual form the same setting is workers_by_key. The key reaches the handler as the partition-key header.

Exclusive: one consumer​

src/main.rs
/// Books payments one after another, in the order they were taken.
#[subscriber(
    PulsarSubscription::new("payments", "ledger").subscription_type(SubscriptionType::Exclusive)
)]
async fn book(payment: &Payment) -> HandlerOutcome {
    tracing::info!(payment.order, payment.amount_cents, "payment booked");
    HandlerOutcome::ack()
}

Exclusive lets one consumer read the subscription, so book sees the payments in the order they were taken. The server refuses a second consumer as busy, and the client keeps asking until the first one leaves or OperationRetries runs out, which by default is never. So a second instance of the service does not start, and reports nothing while it waits. An Exclusive subscription belongs to a service that runs one instance.

Failover: one active consumer​

src/main.rs
/// Invoices orders on one instance at a time; the others stand by.
#[subscriber(
    PulsarSubscription::new("orders", "invoicing").subscription_type(SubscriptionType::Failover)
)]
async fn invoice(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order invoiced");
    HandlerOutcome::ack()
}

Failover accepts every consumer but delivers to one of them. The others are hot standbys: when the active one disconnects, the server hands its messages to the next. Several instances keep the order of an Exclusive subscription without its single point of failure. On a partitioned topic each partition has its own active consumer, so the instances share the partitions.

The app​

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    // A second ledger consumer reports the refusal after three tries instead of waiting.
    let retries = OperationRetries::attempts(nonzero!(3u32));
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650").operation_retries(retries),
        |b| {
            b.include(pack).out(Shipments, Publish).build();
            b.include(track);
            b.include(book);
            b.include(invoice);
        },
    )
}

OperationRetries::attempts bounds how often the client asks again for an operation the server has not accepted. With three attempts, a second consumer of ledger stops the service at startup with an error that names the topic. The bound covers every operation of the connection, the reconnect after a broker restart included, so it trades waiting out an outage for an early error.

Partitioned topics​

A partitioned topic is several topics under one name, and each partition is served on its own. An administrator creates it with its number of partitions:

docker compose exec pulsar bin/pulsar-admin topics create-partitioned-topic \
  persistent://public/default/shipments -p 3

A subscription reads every partition of the topic, and a service reads it with the same descriptor as any other topic. A keyed message goes to the partition its key hashes to, so the messages of one key stay in one partition and in order. Unkeyed messages go to the partitions in turn.

Testing the types​

The tests in the example file run the app in process. There, each subscription gets every message of its topic once, and its type picks the consumer. The in-process Pulsar refuses a second Exclusive consumer at once, so the test fails instead of waiting. Partitions, the key ranges of KeyShared and the consumer a server picks are the server's, and the crate's live tests check them against a running Pulsar.