Перейти к основному содержимому
Нормативная версия документации - английская

Эта страница переведена с английского языковой моделью. При любом расхождении верен английский оригинал.

Повторы и темы dead-letter

Это руководство повторяет запись Kafka сразу или позже, ограничивает число попыток и переносит то, что раз за разом не обрабатывается, в тему dead-letter, на одной теме или на нескольких.

Kafka не придерживает записи и не считает доставки. И то и другое делает фреймворк: точка монтирования объявляет предел попыток и тему dead-letter, а повтор уходит копией записи с числом попыток в заголовке.

Пример - сервис возвратов, который проводит возвраты через процессинг карт.

Подписчик​

Процессинг отказывает в возврате тремя способами, и каждому нужен свой исход. Busy значит, что его лимит запросов исчерпан. Повтор прямо сейчас снова получит отказ, поэтому retry_after ждёт 30 секунд. Unavailable значит, что процессинг не ответил: retry пробует снова сразу. Declined значит, что карточный счёт закрыт. Новая попытка получит тот же ответ, поэтому возврат отбрасывается.

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 завершает запись и по истечении задержки публикует её копию обратно в refunds. Копия несёт заголовок с числом попыток (RETRY_COUNT_HEADER), увеличенным на единицу. Отложенная копия доставляется at-most-once: процесс, который остановился до конца задержки, её теряет. Копия - новая запись в конце темы, поэтому записи после возврата её не ждут.

retry при объявленном пределе тоже становится копией, которая публикуется сразу. Без предела он возвращает запись подписке при Commit::Tracked, как показывает руководство Фиксация и гарантии доставки.

Предел попыток и тема dead-letter​

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 считает все доставки одного возврата, включая первую. Пятая доставка не повторяется: dead_letter публикует её в refunds.dlq с полезной нагрузкой и заголовками в том виде, в каком они пришли. Отброшенный возврат завершается и никуда не уходит. В теме dead-letter лежат только возвраты, у которых кончились попытки. Темы для повторов и для dead-letter - инфраструктура сервиса. Создайте их, как любую другую тему, а фреймворк только публикует в них.

Чтение темы dead-letter​

Тема dead-letter - обычная тема, и review читает её, как любую другую.

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()
}

Заголовок с числом попыток у записи из темы dead-letter хранит число доставок, через которые она прошла, здесь 5. review читает его через ctx.headers() и пишет возврат в журнал, чтобы его посмотрел человек.

Повторы на нескольких темах​

Копия записи публикуется в тему, которую читает подписка. KafkaTopic читает одну тему, поэтому ей больше ничего не нужно. KafkaTopics и KafkaPartitions читают набор, до которого не дотянется ни одна публикация. Поэтому их регистрация называет, куда уходит копия, а без этого сервис не запускается.

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
}

Приложение монтирует process_regional_refund с .out_retry(Publish::default()).transform(ToSourceTopic). ToSourceTopic отправляет каждую копию обратно в тему, из которой пришла её доставка, поэтому повторённый возврат из refunds.us возвращается в refunds.us. Тему он берёт из контекста Kafka, поэтому обработчик принимает ключ контекста, например Ctx<Topic>. В ручной форме тип его контекста - KafkaContext. .out_retry(Publish::default()).to("refunds.retry") вместо этого отправляет все копии в одну вашу тему. Добавьте эту тему в подписку, чтобы копии возвращались. То же правило действует для подписки, которая никогда не повторяет: каждая регистрация на набор тем называет адрес для повторов.

Проверка задержки​

Тесты в файле примера запускают приложение внутри процесса с остановленными часами, поэтому tb.advance(BACKOFF) проматывает 30 секунд мгновенно. Работу с TestApp описывает руководство Тестирование сервиса на Kafka.