Batches

A slice handler reads one fetched delivery plus what librdkafka already holds, up to the size the mount site names.

A handler over &[T] is a batch handler, and batch(n) at the mount site is its largest batch. The size travels down to the consumer's poll, which never hands the body more than n records. librdkafka fetches records ahead in the background, and a batch takes what it already holds without waiting for more. So a batch is often smaller than n, and a quiet topic yields batches of one. One outcome settles every record of the batch. A Vec<HandlerOutcome> settles record i with outcome i. Under Commit::Tracked a record that asks to come back holds the committed position in front of it until it settles; a process that stops first replays the acked records behind it too. A retry_after in a batch becomes a copy published back to the topic after the delay, so the registration names its retry publisher with .out_retry(policy). How much librdkafka queues locally is a consumer property, set through the descriptor's passthrough (queued.max.messages.kbytes, fetch.queue.backoff.ms). workers(n) keeps up to n batches in flight at once. A batch body reads a KafkaBatchContext, which carries the seek handle and no per-record field. In the manual form the body is an impl Handle<[T]>, where Ok(()) acks the batch and Err(outcomes) settles record by record.

Example

Three orders waiting in the consumer's queue arrive in one call.

Macros

examples/kafka_batches.rs

/// Stores a batch of orders in one round-trip; one outcome settles the whole batch.
#[subscriber(
    KafkaTopic::new("orders").group("orders-svc").commit(Commit::Tracked),
    workers(4)
)]
async fn store(orders: &[Order]) -> HandlerOutcome {
    tracing::info!(count = orders.len(), "stored in one round-trip");
    HandlerOutcome::ack()
}

fn app() -> impl App<State = ()> {
    let broker = KafkaBroker::new(["localhost:9092"]);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        // At most 64 orders per call.
        b.include(store.batch(nonzero!(64)));
    })
}

Manual

examples/manual/kafka_batches.rs

/// Stores a batch of orders in one round-trip; `Ok` settles the whole batch.
struct Store;

impl Handle<[Order]> for Store {
    fn handle(
        &self,
        orders: &[Order],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        tracing::info!(count = orders.len(), "stored in one round-trip");
        ready(Ok(()))
    }
}

fn app() -> impl App<State = ()> {
    let broker = KafkaBroker::new(["localhost:9092"]);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        let orders = KafkaTopic::new("orders")
            .group("orders-svc")
            .commit(Commit::Tracked);
        // At most 64 orders per call, up to four calls at once.
        b.include(
            subscriber("orders", Store)
                .batch(nonzero!(64))
                .workers(nonzero!(4))
                .build()
                .map_source(|_| orders),
        );
    })
}

One outcome per record

Only the unsettled payment comes back, as a copy published after the delay.

Macros

examples/kafka_batches.rs

/// One outcome per payment, in slice order: only the unsettled ones come back.
#[subscriber(
    KafkaTopic::new("payments")
        .group("payments-svc")
        .commit(Commit::Tracked)
        .config("queued.max.messages.kbytes", "1024")
)]
async fn reconcile(payments: &[Payment]) -> Vec<HandlerOutcome> {
    payments
        .iter()
        .map(|payment| {
            if payment.settled {
                HandlerOutcome::ack()
            } else {
                HandlerOutcome::retry_after(RECHECK)
            }
        })
        .collect()
}

fn app() -> impl App<State = ()> {
    let broker = KafkaBroker::new(["localhost:9092"]);
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        // A delayed retry is a copy published back to the topic after the delay.
        b.include(reconcile.batch(nonzero!(16)))
            .out_retry(Publish::default());
    })
}

Manual

examples/manual/kafka_batches.rs

/// One outcome per payment, in slice order: only the unsettled ones come back.
struct Reconcile;

impl Handle<[Payment]> for Reconcile {
    fn handle(
        &self,
        payments: &[Payment],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        ready(Err(payments
            .iter()
            .map(|payment| {
                if payment.settled {
                    HandlerOutcome::ack()
                } else {
                    HandlerOutcome::retry_after(RECHECK)
                }
            })
            .collect()))
    }
}

fn app() -> impl App<State = ()> {
    let broker = KafkaBroker::new(["localhost:9092"]);
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        let payments = KafkaTopic::new("payments")
            .group("payments-svc")
            .commit(Commit::Tracked)
            .config("queued.max.messages.kbytes", "1024");
        // A delayed retry is a copy published back to the topic after the delay.
        b.include(
            subscriber("payments", Reconcile)
                .batch(nonzero!(16))
                .build()
                .map_source(|_| payments),
        )
        .out_retry(Publish::default());
    })
}