Skip to main content

Dead letters

dead_letter(topic) names the topic a delivery that reached its cap is republished to, payload and headers as they arrived.

It follows max_attempts(n) at the mount site. When a retry would make the delivery the n-th one, the runtime settles it and publishes it to the dead-letter topic instead. The record keeps its payload, its headers and its record key. RETRY_COUNT_HEADER on it says how many times it was delivered. A cap without a dead-letter topic rejects the spent delivery, which settles its offset, and a warning names the subscription. The dead-letter topic is the service's own infrastructure, an ordinary topic: a subscription of the same service, or of another one, reads it.

Example​

examples/kafka_retries.rs
#[ruststream::app]
fn app() -> impl App<State = AppState> {
    let broker = KafkaBroker::new(["localhost:9092"]);
    RustStream::new(AppInfo::new("payments", "0.1.0"))
        .on_startup(async move |()| {
            let gateway = PaymentGateway {
                endpoint: "https://gateway.internal".into(),
            };
            Ok::<_, Infallible>(AppState { gateway })
        })
        .with_broker(broker, |b| {
            // Five deliveries of one payment, the first included; the fifth failure goes to
            // "payments.dlq" instead, payload and headers as they arrived.
            b.include(charge)
                .max_attempts(nonzero!(5u32))
                .dead_letter("payments.dlq");
            b.include(on_dead_letter);
        })
}

The dead-letter topic is read by an ordinary subscription:

examples/kafka_retries.rs
/// The dead-letter consumer is an ordinary subscription. The framework's retry-count header says
/// how many times the payment was delivered before it was carried here.
#[subscriber(KafkaTopic::new("payments.dlq").group("payments-dlq").start(StartOffset::Earliest))]
async fn on_dead_letter(payment: &Payment, ctx: &mut Context<'_>) -> HandlerOutcome {
    let deliveries = ctx.headers().get_str(RETRY_COUNT_HEADER).unwrap_or("1");
    tracing::error!(payment.id, deliveries, "payment dead-lettered");
    HandlerOutcome::ack()
}