Фиксация и гарантии доставки

Это руководство выбирает, что фиксируют ack и nack в Kafka: собственную фиксацию librdkafka, точное at-least-once с Commit::Tracked или пакеты, которые завершаются поэлементно.

Kafka не завершает отдельные сообщения. Группа консьюмеров фиксирует позицию в каждой партиции, и всё, что ниже неё, считается прочитанным. Обработчик всё равно завершает каждую доставку, поэтому режим Commit подписки решает, как эти завершения превращаются в позицию.

Режим	ack	drop	retry	После падения
Auto (по умолчанию)	ничего	ничего	сообщает, что не поддерживается, и запись не возвращается	хвост обработанных записей читается заново, или сохранённые, но не обработанные записи пропускаются
Tracked	завершает смещение	завершает смещение	запись возвращается	каждая незавершённая запись читается заново
Transactional	завершает смещение, которое фиксируется с транзакцией конвейера	то же, что ack	задерживает окно до срока транзакции	открытое окно читается заново

Пример - сервис выплат: выплаты терять нельзя, а просмотры страниц можно. Третий режим описан в руководстве Транзакции и exactly-once.

At-least-once с Commit::Tracked

pay_out отправляет выплату через банк и отвечает подтверждением. Если банк не ответил, выплата возвращается и обрабатывается снова.

Макросы

src/main.rs

#[derive(Debug, PartialEq, Deserialize)]
struct Payout {
    id: u64,
    amount_cents: u64,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "payouts.done")]
struct Paid {
    payout_id: u64,
}

#[subscriber(KafkaTopic::new("payouts").commit(Commit::Tracked), reply)]
async fn pay_out(payout: &Payout, State(bank): State<Arc<Bank>>) -> Result<Paid, HandlerOutcome> {
    if payout.amount_cents == 0 {
        return Err(HandlerOutcome::drop());
    }
    if bank.0.call().await.is_err() {
        tracing::warn!(payout.id, "bank unavailable, payout comes back");
        return Err(HandlerOutcome::retry());
    }
    tracing::info!(payout.id, payout.amount_cents, "payout sent");
    Ok(Paid {
        payout_id: payout.id,
    })
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Payout {
    id: u64,
    amount_cents: u64,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Paid {
    payout_id: u64,
}

impl OutgoingDestination for Paid {
    type Form = FixedName;

    const DESTINATION: &'static str = "payouts.done";
}

impl MessageHeaders for Paid {
    type Contract = NoHeaders;
}

impl MessageInfo for Paid {
    const NAME: &'static str = "Paid";
}

struct PayOut;

impl Handle<Payout, Paid, (), (), Clients> for PayOut {
    async fn handle(
        &self,
        payout: &Payout,
        _outs: &(),
        ctx: &mut Context<'_, (), Clients>,
    ) -> Result<Paid, HandlerOutcome> {
        if payout.amount_cents == 0 {
            return Err(HandlerOutcome::drop());
        }
        if ctx.state().bank.call().await.is_err() {
            tracing::warn!(payout.id, "bank unavailable, payout comes back");
            return Err(HandlerOutcome::retry());
        }
        tracing::info!(payout.id, payout.amount_cents, "payout sent");
        Ok(Paid {
            payout_id: payout.id,
        })
    }
}

При Commit::Tracked ack поднимает зафиксированную позицию до самой ранней доставки, которая ещё не завершена. Подтверждения от параллельных воркеров приходят в любом порядке, а позиция никогда не обгоняет выплату, которая ещё выполняется. retry возвращает доставку подписке, и та доставляет её снова раньше всего, что выберет дальше. До тех пор её смещение остаётся незавершённым, поэтому позиция за него не уходит. Если процесс остановится раньше, после нового запуска Kafka вернёт ему эту выплату.

Поэтому выплата никогда не теряется, но может быть обработана дважды: до падения и после него. Вызов банка должен принимать повторную выплату, например распознавая её по идентификатору. retry без предела попыток возвращает запись бесконечно. Предел добавляет руководство Повторы и темы dead-letter.

Собственная фиксация librdkafka

count_view считает просмотры страниц в хранилище аналитики. Потерянный просмотр здесь не важен, поэтому подписка остаётся в режиме по умолчанию.

Макросы

src/main.rs

#[derive(Debug, PartialEq, Deserialize)]
struct PageView {
    page: String,
}

#[subscriber(KafkaTopic::new("page-views").group("analytics"))]
async fn count_view(view: &PageView, State(counters): State<Arc<Counters>>) -> HandlerOutcome {
    if counters.0.call().await.is_err() {
        // Under `Commit::Auto` nothing brings the view back; analytics accepts the loss.
        return HandlerOutcome::retry();
    }
    tracing::debug!(page = %view.page, "page view counted");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct PageView {
    page: String,
}

struct CountView;

impl Handle<PageView, (), (), (), Clients> for CountView {
    async fn handle(
        &self,
        view: &PageView,
        _outs: &(),
        ctx: &mut Context<'_, (), Clients>,
    ) -> Result<(), HandlerOutcome> {
        if ctx.state().counters.call().await.is_err() {
            // Under `Commit::Auto` nothing brings the view back; analytics accepts the loss.
            return Err(HandlerOutcome::retry());
        }
        tracing::debug!(page = %view.page, "page view counted");
        Ok(())
    }
}

При Commit::Auto librdkafka сохраняет позицию, когда отдаёт запись приложению, и фиксирует её по таймеру. ack и drop ничего не меняют. retry сообщает, что не поддерживается, и просмотр не возвращается. Этот режим самый дешёвый. Зато при падении можно потерять обработанные записи или пропустить записи, которые были сохранены, но не обработаны.

Пакеты

reconcile принимает строки выписки пакетами до 500 штук и завершает каждую строку отдельно.

Макросы

src/main.rs

#[derive(Debug, PartialEq, Deserialize)]
struct StatementLine {
    payout_id: u64,
    amount_cents: u64,
}

#[subscriber(KafkaTopic::new("statements").group("reconciliation").commit(Commit::Tracked))]
async fn reconcile(lines: &[StatementLine]) -> Vec<HandlerOutcome> {
    lines
        .iter()
        .map(|line| {
            if line.amount_cents == 0 {
                tracing::warn!(line.payout_id, "empty statement line dropped");
                return HandlerOutcome::drop();
            }
            HandlerOutcome::ack()
        })
        .collect()
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct StatementLine {
    payout_id: u64,
    amount_cents: u64,
}

struct Reconcile;

impl Handle<[StatementLine], (), (), (), Clients> for Reconcile {
    fn handle(
        &self,
        lines: &[StatementLine],
        _outs: &(),
        _ctx: &mut Context<'_, (), Clients>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        if lines.iter().all(|line| line.amount_cents > 0) {
            return ready(Ok(()));
        }
        let outcomes = lines
            .iter()
            .map(|line| {
                if line.amount_cents == 0 {
                    tracing::warn!(line.payout_id, "empty statement line dropped");
                    return HandlerOutcome::drop();
                }
                HandlerOutcome::ack()
            })
            .collect();
        ready(Err(outcomes))
    }
}

Пакет - это одна доставка и записи, которые librdkafka уже выбрала, без дополнительного ожидания. Поэтому он часто меньше размера, заданного в точке монтирования. Возвращённый список завершает пакет поэлементно. В ручной форме Ok(()) подтверждает все элементы, а Err(outcomes) завершает каждый по-своему. При Commit::Tracked элемент, который просит повтор, возвращается один, и зафиксированная позиция стоит перед ним, пока он не завершится. Если процесс остановится раньше, подтверждённые элементы после него будут прочитаны снова: это дубликаты, а не потери.

Сколько записей librdkafka держит в очереди, определяет свойство консьюмера, которое задаёт KafkaTopic::config. Когда очередь полна, librdkafka приостанавливает выборку на fetch.queue.backoff.ms (по умолчанию 1000 мс), и большие записи за медленным обработчиком замедляются. .config("fetch.queue.backoff.ms", "10") сокращает эту паузу.

Сторона продюсера

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App<State = Clients> {
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("payouts")
        .producer_config("enable.idempotence", "true")
        .flush_timeout(Duration::from_secs(10));
    RustStream::new(AppInfo::new("payouts", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(broker, |b| {
            b.include(pay_out)
                .out_reply(Publish::default().queue_timeout(Duration::from_secs(5)));
            b.include(count_view);
            b.include(reconcile.batch(nonzero!(500usize)));
        })
}

Вручную

src/main.rs

fn app() -> impl App<State = Clients> {
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("payouts")
        .producer_config("enable.idempotence", "true")
        .flush_timeout(Duration::from_secs(10));
    RustStream::new(AppInfo::new("payouts", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(broker, |b| {
            b.include(
                subscriber("payouts", PayOut)
                    .map_source(|_| KafkaTopic::new("payouts").commit(Commit::Tracked))
                    .reply()
                    .build(),
            )
            .out_reply(Publish::default().queue_timeout(Duration::from_secs(5)));
            b.include(
                subscriber("page-views", CountView)
                    .map_source(|_| KafkaTopic::new("page-views").group("analytics"))
                    .build(),
            );
            b.include(
                subscriber("statements", Reconcile)
                    .batch(nonzero!(500usize))
                    .map_source(|_| {
                        KafkaTopic::new("statements")
                            .group("reconciliation")
                            .commit(Commit::Tracked)
                    })
                    .build(),
            );
        })
}

fn main() -> ExitCode {
    run_main(app)
}

At-least-once от начала до конца требует и стороны продюсера. enable.idempotence делает так, что собственные повторные отправки librdkafka записывают запись один раз. Это и любое другое свойство продюсера, включая acks, задаёт producer_config. Каждая публикация ждёт от кластера отчёта о доставке, поэтому публикация, вернувшая Ok, принята Kafka.

Пока локальная очередь librdkafka полна, публикация ждёт места в ней, и обработчик замедляется до скорости, которую принимает кластер. queue_timeout ограничивает это ожидание и превращает его в ошибку. При остановке брокер дописывает записи, которые ещё в пути, не дольше flush_timeout (по умолчанию 30 секунд). ClosedKafkaBroker::unflushed_records считает записи, которые не успели уйти.

Тесты в файле примера запускают приложение внутри процесса и заменяют банк и хранилище аналитики клиентами, которые один раз отказывают. Kafka внутри процесса фиксирует смещения так же, как кластер, поэтому при Commit::Tracked выплата возвращается, а при Commit::Auto просмотр страницы нет.