KafkaPosition

KafkaPosition names a place in the log: either end, an offset, an offset of one topic, or a timestamp.

Kafka keeps its log, so a subscription can be moved through it, at startup with start_at or from a handler with a seek. KafkaPosition::earliest() is the oldest record each partition retains, and KafkaPosition::latest() is the end of each partition's log. KafkaPosition::timestamp(millis) is, on each partition, the first record stamped at or after that time in epoch milliseconds; a partition with no such record resumes at its end. These three apply to every partition the subscription reads. KafkaPosition::offset(partition, offset) is one partition at an absolute offset, on every topic the subscription reads. KafkaPosition::topic_offset(topic, partition, offset) names the topic too, which a subscription over several topics needs. A delivery's own coordinates are a position as well: Ctx<Position> binds them, and a seek there replays that record. A position on a partition or a topic the subscription does not read is refused with an error.

Example

A new reporting group starts from the first reading stamped at or after the time its configuration names.

Macros

examples/kafka_positions_timestamp.rs

/// Totals meter readings.
#[subscriber("metering.readings")]
async fn total(reading: &Reading) -> HandlerOutcome {
    tracing::info!(reading.id, "totalled");
    HandlerOutcome::ack()
}

/// The service, reading from the first record stamped at or after `since` (epoch milliseconds).
fn app(servers: &str, group: &str, since: i64) -> impl App<State = ()> {
    let broker = KafkaBroker::new([servers]).default_group(group);
    RustStream::new(AppInfo::new("metering", "0.1.0")).with_broker(broker, |b| {
        b.include(total.start_at(KafkaPosition::timestamp(since)));
    })
}

Manual

examples/manual/kafka_positions_timestamp.rs

/// Totals meter readings.
struct Total;

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

/// The service, reading from the first record stamped at or after `since` (epoch milliseconds).
fn app(servers: &str, group: &str, since: i64) -> impl App<State = ()> {
    let broker = KafkaBroker::new([servers]).default_group(group);
    RustStream::new(AppInfo::new("metering", "0.1.0")).with_broker(broker, |b| {
        b.include(
            subscriber("metering.readings", Total)
                .start_at(KafkaPosition::timestamp(since))
                .build(),
        );
    })
}