Replay and seeking

This guide reads a topic again from the start or from a moment, and moves a running subscription from inside its handler.

Kafka keeps the log after a consumer has read it. A service can read it again: to rebuild a read model, to backfill a new consumer, or to skip records it cannot use. A subscription takes its start position at the mount site, and a handler moves its own subscription through the delivery's context.

The read models

The service keeps two read models in memory: the balance of every account and the spend of every card. Each posting carries the ledger's revision. A correction of past postings is published as a new revision, and a read model that meets one rebuilds itself from the start of the log.

Macros
Manual

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>>,
}

Start from the beginning

The balances live in memory, so after a restart they are built again from the whole log.

Macros
Manual

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()) opens the subscription at the start of the log on every start, whatever the group balances committed before. StartOffset on the descriptor differs here: it applies only while the group has no committed offset. The position lands on the partitions the group assigns to this consumer, before their first record.

KafkaPosition has five positions:

Position	Where the subscription resumes
`earliest()`	the oldest record Kafka keeps
`latest()`	the end of the log, so only new records are read
`offset(partition, n)`	offset `n` of one partition
`topic_offset(topic, partition, n)`	offset `n` of one partition of one topic, for a subscription that reads several topics
`timestamp(ms)`	the first record stamped at or after `ms`, in every partition

A start position that names none of the subscription's partitions fails the start. Under Assignment::CooperativeSticky the first assignment may be only part of the group's share, and partitions that arrive in a later increment are not moved.

Seek from a handler

The same handler rebuilds the balances when a posting of a newer revision arrives. Ctx<SeekHandle> gives the handler the KafkaSeeker of its own subscription. seek moves the subscription to the start of the log, and the handler reads every posting again. The posting that started the rebuild comes back as well, and this time the projection applies it.

A seek takes any of the positions above. Ctx<Position> holds the delivery's own position: a seek there reads the record again, and everything after it. Ctx<Partition> and Ctx<Offset> give the coordinates for an offset(..) position.

Seek from a batch handler

Card postings arrive in volume, so the spend is built in batches of up to 500.

Macros
Manual

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

A batch handler declares KafkaBatchContext instead of the per-delivery context. It carries the same SeekHandle key and no position, because a batch spans many records. The subscription moves once the whole batch has settled.

Start from a moment

The payout review reads the payouts of the last seven days on every start.

Macros
Manual

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

The moment is known only when the service starts, so the mount site names the position instead of the attribute.

Macros
Manual

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 takes milliseconds since the epoch, the unit Kafka stamps records with. Kafka resolves the moment in each partition to the first record stamped at or after it. A partition with no such record resumes at its end.

What a seek does to the group

A seek moves this consumer, not the group. It moves the partitions this member holds and commits nothing for anyone. Other members of the group keep reading where they are. The next rebalance discards the seek: whoever gets a partition then resumes from the group's committed offset. A start_at position applies again at every start.

Seeking under Commit::Tracked

All three subscriptions commit with Commit::Tracked (Commits and delivery guarantees). A seek clears the tracked position of every partition it moves. A delivery pulled before the seek settles into nothing, so the ack of the posting that started the rebuild commits nothing. A later commit therefore never passes a record that the seek replayed and no handler has handled yet.

The read models​

Start from the beginning​

Seek from a handler​

Seek from a batch handler​

Start from a moment​

What a seek does to the group​

Seeking under Commit::Tracked​