start_at

start_at opens a subscription at a position on every startup, whatever the group committed.

It is a clause of the attribute, start_at(KafkaPosition::earliest()), or .start_at(position) among the mount-site settings. The position lands on the partitions the subscription opened with, ahead of their first record. A subscription that opened with no partitions takes the position on the first assignment its group hands over. Under Assignment::CooperativeSticky that first assignment may be part of the group's share, and partitions that arrive in a later increment start from the committed offset. A position that names none of the partitions the subscription opened with stops the service at startup. It suits a read model rebuilt from the retained log on each boot. A group's start offset is different: it applies only where the group has committed nothing.

Example

The audit projection is rebuilt from the oldest retained entry on every startup, although its group committed everything it read before.

Macros

examples/kafka_seek.rs

/// Rebuilds the audit projection from the oldest retained entry on every startup.
#[subscriber(
    KafkaTopic::new("ledger.audit").commit(Commit::Tracked),
    start_at(KafkaPosition::earliest())
)]
async fn replay(entry: &AuditEntry) -> HandlerOutcome {
    tracing::info!(entry.id, "projected");
    HandlerOutcome::ack()
}

Manual

examples/manual/kafka_seek.rs

/// Projects every audit entry it reads.
struct Replay;

impl Handle<AuditEntry> for Replay {
    fn handle(
        &self,
        entry: &AuditEntry,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(entry.id, "projected");
        ready(Ok(()))
    }
}

fn app(servers: &str, group: &str) -> impl App<State = ()> {
    let broker = KafkaBroker::new([servers]).default_group(group);
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(broker, |b| {
        let audit = KafkaTopic::new("ledger.audit").commit(Commit::Tracked);
        // The projection is rebuilt from the oldest retained entry on every startup.
        b.include(
            subscriber("ledger.audit", Replay)
                .build()
                .map_source(|_| audit)
                .start_at(KafkaPosition::earliest()),
        );
    })
}