Skip to main content

OperationRetries

OperationRetries bounds how long the client keeps retrying an operation a server has not accepted, which is unbounded by default.

A server sometimes answers "not now": a broker is still starting, a topic is moving between brokers, a subscription is taken. The client then tries again, by default for ever and five seconds apart. That is what waits out a broker restart.

attempts(n) gives up after n tries and reports what the server answered. One try makes the first answer final. delay sets the wait between tries, five seconds by default, and timeout the time one try has to be answered, thirty seconds by default. PulsarBroker::operation_retries applies the bound.

The bound covers every retried operation of the connection: topic lookups, opening producers and consumers, and a consumer's reconnect after a disconnect. A short bound turns a long outage into a subscription that gives up.

The case that needs a bound is an Exclusive subscription another consumer holds. Unbounded, the second consumer waits for the first to leave, and the service does not start. Bounded, startup fails with PulsarError::Subscribe, naming the topic. In a TestApp test the in-process transport refuses a second Exclusive consumer at once, so the bound shows only against a server.

Connecting to a production cluster bounds the retries of a service and tests the configuration.

Example​

The ledger has one writer at a time. A second instance of the service tries three times, a second apart, and then fails its startup instead of waiting.

examples/pulsar_broker_retries.rs
/// The cluster at `url`; an operation the server keeps refusing fails after three tries.
fn broker(url: &str) -> PulsarBroker {
    PulsarBroker::new(url)
        .operation_retries(OperationRetries::attempts(nonzero!(3u32)).delay(Duration::from_secs(1)))
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(broker("pulsar://localhost:6650"))
}