Повторное чтение и перемотка

Это руководство читает тему заново с начала или с заданного момента и перематывает работающую подписку из её обработчика.

Kafka хранит журнал и после того, как консьюмер его прочитал. Сервис может прочитать его снова: чтобы перестроить модель чтения, заполнить данными новый консьюмер или пропустить записи, которые ему не годятся. Подписка получает начальную позицию в точке монтирования, а обработчик перематывает свою подписку через контекст доставки.

Модели чтения

Сервис держит в памяти две модели чтения: баланс каждого счёта и расходы по каждой карте. Каждая проводка несёт ревизию реестра. Исправление прошлых проводок публикуется новой ревизией, и модель чтения, которая её встречает, перестраивается с начала журнала.

Макросы

src/main.rs

/// One posting of the ledger. `revision` is the ledger's revision: a correction of past
/// postings is published as a new revision, and a read model meeting one rebuilds itself.
#[derive(Debug, PartialEq, Deserialize)]
struct Entry {
    id: u64,
    account: String,
    amount_cents: i64,
    revision: u32,
}

/// A read model kept in memory: the balance of every account.
#[derive(Debug, Default)]
struct Projection {
    revision: u32,
    balances: HashMap<String, i64>,
}

impl Projection {
    /// Applies `entry`, or clears the model and answers `false` when the entry belongs to a newer
    /// revision, which the model has to be rebuilt for from the start of the log.
    fn apply(&mut self, entry: &Entry) -> bool {
        if entry.revision > self.revision {
            self.revision = entry.revision;
            self.balances.clear();
            return false;
        }
        *self.balances.entry(entry.account.clone()).or_default() += entry.amount_cents;
        true
    }
}

/// The service's state: one read model per subscription that builds one.
#[derive(Clone, Default)]
struct Projections {
    balances: Arc<Mutex<Projection>>,
    spend: Arc<Mutex<Projection>>,
}

Вручную

src/main.rs

/// One posting of the ledger. `revision` is the ledger's revision: a correction of past
/// postings is published as a new revision, and a read model meeting one rebuilds itself.
#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Entry {
    id: u64,
    account: String,
    amount_cents: i64,
    revision: u32,
}

/// A read model kept in memory: the balance of every account.
#[derive(Debug, Default)]
struct Projection {
    revision: u32,
    balances: HashMap<String, i64>,
}

impl Projection {
    /// Applies `entry`, or clears the model and answers `false` when the entry belongs to a newer
    /// revision, which the model has to be rebuilt for from the start of the log.
    fn apply(&mut self, entry: &Entry) -> bool {
        if entry.revision > self.revision {
            self.revision = entry.revision;
            self.balances.clear();
            return false;
        }
        *self.balances.entry(entry.account.clone()).or_default() += entry.amount_cents;
        true
    }
}

/// The service's state: one read model per subscription that builds one.
#[derive(Clone, Default)]
struct Projections {
    balances: Arc<Mutex<Projection>>,
    spend: Arc<Mutex<Projection>>,
}

Чтение с начала

Балансы живут в памяти, поэтому после перезапуска они строятся заново по всему журналу.

Макросы

src/main.rs

#[subscriber(
    KafkaTopic::new("ledger").group("balances").commit(Commit::Tracked),
    start_at(KafkaPosition::earliest())
)]
async fn project_balance(
    entry: &Entry,
    ctx: &mut Context<'_, KafkaContext, Projections>,
    Ctx(seeker): Ctx<SeekHandle>,
) -> HandlerOutcome {
    let applied = ctx
        .state()
        .balances
        .lock()
        .unwrap_or_else(PoisonError::into_inner)
        .apply(entry);
    if applied {
        tracing::info!(entry.id, "balance updated");
        return HandlerOutcome::ack();
    }
    tracing::info!(
        entry.id,
        entry.revision,
        "new revision, rebuilding the balances"
    );
    if seeker.seek(KafkaPosition::earliest()).await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Declaring `KafkaContext` is what puts the seek handle in the context.
struct ProjectBalance;

impl Handle<Entry, (), (), KafkaContext, Projections> for ProjectBalance {
    async fn handle(
        &self,
        entry: &Entry,
        _outs: &(),
        ctx: &mut Context<'_, KafkaContext, Projections>,
    ) -> Result<(), HandlerOutcome> {
        let applied = ctx
            .state()
            .balances
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .apply(entry);
        if applied {
            tracing::info!(entry.id, "balance updated");
            return Ok(());
        }
        tracing::info!(
            entry.id,
            entry.revision,
            "new revision, rebuilding the balances"
        );
        if ctx
            .context(SeekHandle)
            .seek(KafkaPosition::earliest())
            .await
            .is_err()
        {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

start_at(KafkaPosition::earliest()) открывает подписку с начала журнала при каждом запуске, что бы группа balances ни зафиксировала раньше. StartOffset в дескрипторе работает иначе: он действует, только пока у группы нет зафиксированного смещения. Позиция ставится на партиции, которые группа назначает этому консьюмеру, до их первой записи.

У KafkaPosition пять позиций:

Позиция	Откуда продолжает подписка
earliest()	с самой старой записи, которую хранит Kafka
latest()	с конца журнала, поэтому читаются только новые записи
offset(partition, n)	со смещения n одной партиции
topic_offset(topic, partition, n)	со смещения n одной партиции одной темы, для подписки на несколько тем
timestamp(ms)	с первой записи с отметкой времени не раньше ms, в каждой партиции

Если начальная позиция не называет ни одной партиции подписки, запуск завершается ошибкой. При Assignment::CooperativeSticky первое назначение может быть лишь частью доли группы, а партиции, которые приходят следующими порциями, не перематываются.

Перемотка из обработчика

Тот же обработчик перестраивает балансы, когда приходит проводка более новой ревизии. Ctx<SeekHandle> даёт обработчику KafkaSeeker его собственной подписки. seek перематывает подписку на начало журнала, и обработчик читает все проводки заново. Проводка, с которой началась перестройка, тоже возвращается, и на этот раз проекция её применяет.

Перемотка принимает любую из позиций выше. Ctx<Position> хранит собственную позицию доставки: перемотка на неё читает запись заново, а за ней и всё последующее. Ctx<Partition> и Ctx<Offset> дают координаты для позиции offset(..).

Перемотка из пакетного обработчика

Проводки по картам идут потоком, поэтому расходы строятся пакетами до 500 штук.

Макросы

src/main.rs

#[subscriber(KafkaTopic::new("card-entries")
        .group("card-spend")
        .commit(Commit::Tracked))]
async fn project_spend(
    entries: &[Entry],
    ctx: &mut Context<'_, KafkaBatchContext, Projections>,
) -> HandlerOutcome {
    let applied = {
        let mut spend = ctx
            .state()
            .spend
            .lock()
            .unwrap_or_else(PoisonError::into_inner);
        entries.iter().all(|entry| spend.apply(entry))
    };
    if applied {
        tracing::info!(entries = entries.len(), "card spend updated");
        return HandlerOutcome::ack();
    }
    tracing::info!("new revision, rebuilding the card spend");
    if ctx
        .context(SeekHandle)
        .seek(KafkaPosition::earliest())
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// A batch body declares `KafkaBatchContext`, which carries the same seek handle.
struct ProjectSpend;

impl Handle<[Entry], (), (), KafkaBatchContext, Projections> for ProjectSpend {
    async fn handle(
        &self,
        entries: &[Entry],
        _outs: &(),
        ctx: &mut Context<'_, KafkaBatchContext, Projections>,
    ) -> Result<(), Vec<HandlerOutcome>> {
        let applied = {
            let mut spend = ctx
                .state()
                .spend
                .lock()
                .unwrap_or_else(PoisonError::into_inner);
            entries.iter().all(|entry| spend.apply(entry))
        };
        if applied {
            tracing::info!(entries = entries.len(), "card spend updated");
            return Ok(());
        }
        tracing::info!("new revision, rebuilding the card spend");
        if ctx
            .context(SeekHandle)
            .seek(KafkaPosition::earliest())
            .await
            .is_err()
        {
            return Err(entries.iter().map(|_| HandlerOutcome::retry()).collect());
        }
        Ok(())
    }
}

Пакетный обработчик объявляет KafkaBatchContext вместо контекста одной доставки. В нём тот же ключ SeekHandle и нет позиции, потому что пакет охватывает много записей. Подписка перематывается, когда завершён весь пакет.

Чтение с заданного момента

Проверка выплат при каждом запуске читает выплаты за последние семь дней.

Макросы

src/main.rs

/// A payout as the payments service publishes it.
#[derive(Debug, PartialEq, Deserialize)]
struct Payout {
    id: u64,
    amount_cents: i64,
}

#[subscriber(KafkaTopic::new("payouts")
        .group("payout-review")
        .commit(Commit::Tracked))]
async fn review_payout(payout: &Payout) -> HandlerOutcome {
    tracing::info!(payout.id, payout.amount_cents, "payout reviewed");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// A payout as the payments service publishes it.
#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Payout {
    id: u64,
    amount_cents: i64,
}

/// The body reads no context, but runs in an app whose state is `Projections`.
struct ReviewPayout;

impl Handle<Payout, (), (), (), Projections> for ReviewPayout {
    fn handle(
        &self,
        payout: &Payout,
        _outs: &(),
        _ctx: &mut Context<'_, (), Projections>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(payout.id, payout.amount_cents, "payout reviewed");
        ready(Ok(()))
    }
}

Момент известен только при запуске сервиса, поэтому позицию задаёт точка монтирования, а не атрибут.

Макросы

src/main.rs

/// Milliseconds since the epoch, the unit Kafka stamps records with.
fn epoch_millis(at: SystemTime) -> i64 {
    at.duration_since(UNIX_EPOCH).map_or(0, |since| {
        i64::try_from(since.as_millis()).unwrap_or(i64::MAX)
    })
}

fn service(broker: KafkaBroker, review_from: SystemTime) -> impl App<State = Projections> {
    RustStream::new(AppInfo::new("ledger", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Projections::default()))
        .with_broker(broker, |b| {
            b.include(project_balance);
            b.include(project_spend.batch(nonzero!(500usize)));
            b.include(review_payout.start_at(KafkaPosition::timestamp(epoch_millis(review_from))));
        })
}

/// The review reads the payouts of the last seven days on every start.
const REVIEW_WINDOW: Duration = Duration::from_hours(7 * 24);

#[ruststream::app]
fn app() -> impl App<State = Projections> {
    let review_from = SystemTime::now() - REVIEW_WINDOW;
    service(KafkaBroker::new(["localhost:9092"]), review_from)
}

Вручную

src/main.rs

/// Milliseconds since the epoch, the unit Kafka stamps records with.
fn epoch_millis(at: SystemTime) -> i64 {
    at.duration_since(UNIX_EPOCH).map_or(0, |since| {
        i64::try_from(since.as_millis()).unwrap_or(i64::MAX)
    })
}

fn service(broker: KafkaBroker, review_from: SystemTime) -> impl App<State = Projections> {
    RustStream::new(AppInfo::new("ledger", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Projections::default()))
        .with_broker(broker, |b| {
            b.include(
                subscriber("ledger", ProjectBalance)
                    .map_source(|_| {
                        KafkaTopic::new("ledger")
                            .group("balances")
                            .commit(Commit::Tracked)
                    })
                    .start_at(KafkaPosition::earliest())
                    .build(),
            );
            b.include(
                subscriber("card-entries", ProjectSpend)
                    .map_source(|_| {
                        KafkaTopic::new("card-entries")
                            .group("card-spend")
                            .commit(Commit::Tracked)
                    })
                    .batch(nonzero!(500usize))
                    .build(),
            );
            b.include(
                subscriber("payouts", ReviewPayout)
                    .map_source(|_| {
                        KafkaTopic::new("payouts")
                            .group("payout-review")
                            .commit(Commit::Tracked)
                    })
                    .start_at(KafkaPosition::timestamp(epoch_millis(review_from)))
                    .build(),
            );
        })
}

/// The review reads the payouts of the last seven days on every start.
const REVIEW_WINDOW: Duration = Duration::from_hours(7 * 24);

fn app() -> impl App<State = Projections> {
    let review_from = SystemTime::now() - REVIEW_WINDOW;
    service(KafkaBroker::new(["localhost:9092"]), review_from)
}

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

KafkaPosition::timestamp принимает миллисекунды от начала эпохи: в этих единицах Kafka ставит отметки времени на записи. Kafka находит момент в каждой партиции по первой записи с отметкой не раньше него. Партиция без такой записи продолжает со своего конца.

Что перемотка делает с группой

Перемотка двигает этот консьюмер, а не группу. Она переносит партиции, которые держит этот участник, и ни за кого ничего не фиксирует. Остальные участники группы продолжают читать с того места, где находятся. Следующая перебалансировка отменяет перемотку: тот, кому достанется партиция, продолжит с зафиксированного смещения группы. Позиция start_at применяется заново при каждом запуске.

Перемотка при Commit::Tracked

Все три подписки фиксируют смещения с Commit::Tracked (см. Фиксация и гарантии доставки). Перемотка сбрасывает отслеживаемую позицию каждой партиции, которую переносит. Доставка, полученная до перемотки, завершается впустую, поэтому подтверждение проводки, с которой началась перестройка, ничего не фиксирует. Так следующая фиксация никогда не обгоняет запись, которую перемотка прочитала заново и которую ещё не обработал ни один обработчик.