StartOffset

StartOffset says where a group starts on a partition with no committed offset: committed reset, earliest or latest.

A group resumes from its committed offset whenever it has a valid one. StartOffset applies only where it has none: before the group's first commit on a partition, and after retention deleted the record a committed offset pointed at. The setting is librdkafka's auto.offset.reset, written on the descriptor with start. StartOffset::Committed, the default, leaves the choice to librdkafka, which resets to the end of the log. StartOffset::Earliest starts at the oldest record the topic retains. StartOffset::Latest starts at the end of the log: only records published after the subscription opened arrive. The second case matters for a group that sat idle longer than the topic's retention: with the default it skips to the end instead of reading what is left. A position that applies on every startup, whatever the group committed, is start_at.

Example

A new archive group reads the invoices the topic held before the group existed.

Macros

examples/kafka_subscriptions_start_offset.rs

/// Archives every invoice: a new archive group starts at the oldest record the topic retains.
#[subscriber(KafkaTopic::new("ledger.invoices").start(StartOffset::Earliest))]
async fn archive(invoice: &Invoice) -> HandlerOutcome {
    tracing::info!(invoice.id, "archived");
    HandlerOutcome::ack()
}

Manual

examples/manual/kafka_subscriptions_start_offset.rs

/// Archives every invoice it reads.
struct Archive;

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

fn app(servers: &str, group: &str) -> impl App {
    let broker = KafkaBroker::new([servers]).default_group(group);
    RustStream::new(AppInfo::new("archive", "0.1.0")).with_broker(broker, |b| {
        // A new archive group starts at the oldest record the topic retains.
        let invoices = KafkaTopic::new("ledger.invoices").start(StartOffset::Earliest);
        b.include(
            subscriber("ledger.invoices", Archive)
                .build()
                .map_source(|_| invoices),
        );
    })
}