Retries

retry_after(delay) asks for a later delivery: the runtime republishes a copy to the subscription with its delivery count in a header.

Kafka holds no record back and counts no deliveries, so the runtime does both. It settles the delivery, waits out the delay and publishes a copy of the record, payload and headers, to the topic the subscription reads. The copy carries RETRY_COUNT_HEADER incremented, and the handler reads it like any other header. The copy waits in the service's memory, so it is at-most-once over the delay: a process that exits before the timer fires loses it. max_attempts(n) caps the deliveries of one record at the mount site, the first one included. Under a cap every retry, an immediate one too, travels as such a copy, because the count lives in the header. The delivery that reaches the cap goes to the dead-letter topic. The copies leave through the Retry slot, which out_retry replaces. A subscription over several topics names where its copies go, as ToSourceTopic shows.

Example

Macros
Manual

examples/kafka_retries.rs
/// A transient failure asks for a later delivery. Kafka cannot hold a record back, so the
/// runtime waits out the delay and publishes a copy to the topic this subscription reads, with
/// the retry-count header incremented.
#[subscriber(KafkaTopic::new("payments").group("payments-svc").commit(Commit::Tracked))]
async fn charge(payment: &Payment, State(gateway): State<PaymentGateway>) -> HandlerOutcome {
    if payment.amount_cents <= 0 {
        // Malformed input is not worth retrying: dropping it settles the offset.
        return HandlerOutcome::drop();
    }
    match gateway.charge(payment).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(payment.id, %err, "charge failed, retrying");
            HandlerOutcome::retry_after(RECHECK)
        }
    }
}

examples/manual/kafka_retries.rs
/// A transient failure asks for a later delivery. Kafka cannot hold a record back, so the
/// runtime waits out the delay and publishes a copy to the topic this subscription reads, with
/// the retry-count header incremented.
struct Charge;

impl Handle<Payment, (), (), (), AppState> for Charge {
    async fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        ctx: &mut Context<'_, (), AppState>,
    ) -> Result<(), HandlerOutcome> {
        if payment.amount_cents <= 0 {
            // Malformed input is not worth retrying: dropping it settles the offset.
            return Err(HandlerOutcome::drop());
        }
        if let Err(err) = ctx.state().gateway.charge(payment).await {
            tracing::warn!(payment.id, %err, "charge failed, retrying");
            return Err(HandlerOutcome::retry_after(RECHECK));
        }
        Ok(())
    }
}

fn payments() -> KafkaTopic {
    KafkaTopic::new("payments")
        .group("payments-svc")
        .commit(Commit::Tracked)
}

Example​

Example