Skip to main content

Retries and dead-letter topics

This guide retries a Kafka record now or later, caps the attempts and moves what keeps failing to a dead-letter topic, on one topic or several.

Kafka does not hold a record back and does not count deliveries. The framework does both: the mount site declares the cap and the dead-letter topic, and a retry travels as a copy of the record with a retry count in a header.

The example is a refunds service that sends refunds through a card processor.

The subscriber​

The processor refuses a refund in three ways, and each needs a different outcome. Busy means its rate limit is used up: a retry right now would be refused again, so retry_after waits 30 seconds. Unavailable means it did not answer: retry tries again at once. Declined means the card account is closed, and another try gets the same answer, so the refund is dropped.

src/main.rs
/// How long a busy processor asks its callers to wait.
const BACKOFF: Duration = Duration::from_secs(30);

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Refund {
    id: u64,
    card: String,
    amount_cents: u64,
}

#[subscriber(KafkaTopic::new("refunds").commit(Commit::Tracked))]
async fn process_refund(
    refund: &Refund,
    State(processor): State<Arc<Processor>>,
) -> HandlerOutcome {
    settle(&processor, refund).await
}

async fn settle(processor: &Processor, refund: &Refund) -> HandlerOutcome {
    match processor.refund(refund).await {
        Ok(()) => HandlerOutcome::ack(),
        // Past the rate limit: a delivery right now would be refused again.
        Err(Refusal::Busy) => HandlerOutcome::retry_after(BACKOFF),
        // No answer: try again at once, up to the attempt cap.
        Err(Refusal::Unavailable) => HandlerOutcome::retry(),
        Err(Refusal::Declined) => {
            tracing::warn!(refund.id, "card declined, refund dropped");
            HandlerOutcome::drop()
        }
    }
}

retry_after settles the record and publishes a copy back to refunds once the delay has passed. The copy carries the retry-count header (RETRY_COUNT_HEADER), raised by one. The delayed copy is at-most-once: a process that stops before the delay ends loses it. The copy is a new record at the end of the topic, so the records behind the refund do not wait for it.

retry under a declared cap also becomes a copy, published at once. Without a cap it hands the record back to the subscription under Commit::Tracked, as Commits and delivery guarantees shows.

The cap and the dead-letter topic​

src/main.rs
#[ruststream::app]
fn app() -> impl App<State = Clients> {
    RustStream::new(AppInfo::new("refunds", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(
            KafkaBroker::new(["localhost:9092"]).default_group("refunds"),
            |b| {
                // Five deliveries of one refund, the first included; then the dead-letter topic.
                b.include(process_refund)
                    .max_attempts(nonzero!(5u32))
                    .dead_letter("refunds.dlq");
                b.include(review);
                // A set of topics names where a copy goes: back to the topic it came from.
                b.include(process_regional_refund)
                    .max_attempts(nonzero!(5u32))
                    .dead_letter("refunds.dlq")
                    .out_retry(Publish::default())
                    .transform(ToSourceTopic);
            },
        )
}

max_attempts counts every delivery of one refund, the first included. The fifth delivery is not retried: dead_letter publishes it to refunds.dlq, payload and headers as they arrived. A dropped refund settles and goes nowhere; the dead-letter topic holds only the refunds that ran out of attempts. Retry and dead-letter topics are the service's infrastructure: create them like any other topic, and the framework only publishes to them.

Reading the dead-letter topic​

The dead-letter topic is an ordinary topic, and review reads it like any other.

src/main.rs
#[subscriber(
    KafkaTopic::new("refunds.dlq")
        .group("refunds-review")
        .start(StartOffset::Earliest)
)]
async fn review(refund: &Refund, ctx: &mut Context<'_>) -> HandlerOutcome {
    let deliveries = ctx.headers().get_str(RETRY_COUNT_HEADER).unwrap_or("1");
    tracing::error!(refund.id, deliveries, "refund needs a person to look at it");
    HandlerOutcome::ack()
}

The retry-count header of a dead-lettered record holds the number of deliveries it went through, 5 here. review reads it with ctx.headers() and logs the refund for a person to look at.

Retries on several topics​

A copy of a record is published to the topic the subscription reads. KafkaTopic reads one topic, so it needs nothing more. KafkaTopics and KafkaPartitions read a set that no single publish reaches, so their registration names where a copy goes, and the service refuses to start without it.

src/main.rs
#[subscriber(
    KafkaTopics::new(["refunds.eu", "refunds.us"])
        .group("regional-refunds")
        .commit(Commit::Tracked)
)]
async fn process_regional_refund(
    refund: &Refund,
    Ctx(topic): Ctx<Topic>,
    State(processor): State<Arc<Processor>>,
) -> HandlerOutcome {
    tracing::debug!(refund.id, %topic, "regional refund");
    settle(&processor, refund).await
}

The app mounts process_regional_refund with .out_retry(Publish::default()).transform(ToSourceTopic). ToSourceTopic sends each copy back to the topic its delivery came from, so a retried refund from refunds.us comes back on refunds.us. It reads the topic from the Kafka context, so the handler takes a context key such as Ctx<Topic>; in the manual form its context type is KafkaContext. .out_retry(Publish::default()).to("refunds.retry") sends every copy to one topic of your own instead; list that topic in the subscription so the copies come back. The same rule holds for a subscription that never retries: every registration over a set of topics names a retry destination.

Testing the delay​

The tests in the example file run the app in process with a paused clock, so tb.advance(BACKOFF) lets the 30 seconds pass at once. Testing a Kafka service covers the harness.