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

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

Повторы и dead-letter

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

Pulsar сам доставляет сообщение повторно, поэтому повтор ничего не публикует. Точка монтирования объявляет предел и топик dead-letter, а применяет их клиент Pulsar.

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

Подписчик​

Процессинг отказывает в возврате тремя способами, и каждому нужен свой исход. 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)]
#[outgoing(name = "refunds")]
struct Refund {
    id: u64,
    card: String,
    amount_cents: u64,
}

/// A refund left unsettled for two minutes comes back; a wait must be shorter than that.
#[subscriber(PulsarSubscription::new("refunds", "refunds").ack_timeout(Duration::from_secs(120)))]
async fn process_refund(
    refund: &Refund,
    State(processor): State<Arc<Processor>>,
) -> 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 отрицательно подтверждает возврат, и брокер сразу доставляет его снова. retry_after держит возврат в сервисе неподтверждённым в течение задержки, а затем отрицательно подтверждает его. У отрицательного подтверждения Pulsar нет собственной задержки, поэтому ожидание идёт в процессе. Процесс, который остановился во время ожидания, теряет ожидание, а не возврат: брокер доставит возврат снова после тайм-аута подтверждения или сразу, когда консьюмер отключится.

Задержка должна быть короче тайм-аута подтверждения подписки, иначе тайм-аут вернул бы возврат раньше. Более долгая задержка отвергается, когда обработчик её возвращает, с ошибкой, которая называет оба значения. Подписка без тайм-аута подтверждения принимает любую задержку.

Предел и топик 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(PulsarBroker::new("pulsar://localhost:6650"), |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);
        })
}

max_attempts считает все доставки одного возврата, включая первую. dead_letter называет, куда уходит возврат, когда доставки исчерпаны. Эта пара становится собственной политикой dead-letter консьюмера. Клиент Pulsar считает повторные доставки, публикует пятый неудачный возврат в refunds-dlq и подтверждает оригинал. retry, retry_after и истёкший тайм-аут подтверждения засчитываются как повторная доставка. Отброшенный возврат завершается и никуда не уходит. В топике dead-letter лежат только возвраты, у которых кончились попытки.

Указывайте оба шага или ни одного: предел без адреса или адрес без предела останавливает сервис при запуске, и ошибка называет недостающий шаг. Брокер считает повторные доставки только на подписках Shared и KeyShared, поэтому предел задаётся на них. Предел на подписке Exclusive или Failover никогда не был бы достигнут, и такая регистрация останавливает сервис при запуске с ошибкой, которая называет тип. Обработчик, смонтированный по короткому имени топика, принимает то же объявление на подписке брокера по умолчанию.

Чтение топика dead-letter​

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

src/main.rs
/// Reads the refunds that ran out of attempts.
#[subscriber(PulsarSubscription::new("refunds-dlq", "refunds-review"))]
async fn review(refund: &Refund) -> HandlerOutcome {
    tracing::error!(refund.id, "refund needs a person to look at it");
    HandlerOutcome::ack()
}

Возврат в топике dead-letter сохраняет полезную нагрузку и заголовки, с которыми пришёл. Клиент добавляет два свойства: REAL_TOPIC, топик, из которого пришёл возврат, и ORIGIN_MESSAGE_ID, его позицию там. review пишет возврат в журнал, чтобы его посмотрел человек.

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

Тесты в файле примера запускают приложение внутри процесса с TestApp::with_state, который передаёт обработчикам намеренно отказывающий процессинг. Тест задержки начинается с остановленными часами, поэтому tb.advance(BACKOFF) проматывает 30 секунд мгновенно. Pulsar внутри процесса считает доставки и переносит возврат с исчерпанными попытками в refunds-dlq так же, как клиент. Работу с TestApp описывает руководство Тестирование сервиса на Pulsar.