Skip to main content

Batches

A slice handler gets up to batch(n) records; n becomes the GetRecords limit, and a short batch goes out 50 ms after its first record.

A handler over &[T] is a batch handler, and batch(n) at the mount site is its largest batch. Every shard reader asks GetRecords for at most n records, so one read never fetches more than one batch holds; the service caps a read at 10,000 records. A batch that is still filling goes to the handler 50 ms after its first record, so on a quiet stream a batch is often smaller than n. One batch mixes the records of every shard this instance reads. One outcome settles every record of the batch. A Vec<HandlerOutcome> settles record i with outcome i, and each record moves the checkpoint of its own shard. A retry_after in a batch becomes a copy published back to the stream after the delay. A small batch means small reads against the same five reads a second per shard, which poll_interval paces. A batch body reads a KinesisBatchContext, which carries the seek handle and no position of a single record. In the manual form the body is an impl Handle<[T]>, where Ok(()) acks the batch and Err(outcomes) settles record by record.

Example​

Orders are stored up to 500 at a time; refunds are reviewed up to 100 at a time.

examples/kinesis_batches.rs
/// Stores a batch of orders in one round-trip and settles the whole batch with one outcome.
#[subscriber(KinesisStream::new("orders"))]
async fn digest(batch: &[Order]) -> HandlerOutcome {
    tracing::info!(count = batch.len(), "orders stored");
    HandlerOutcome::ack()
}

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        // The framework's step first, this crate's read setting after it.
        b.include(
            digest
                .batch(nonzero!(500))
                .poll_interval(Duration::from_millis(500)),
        );
        b.include(review.batch(nonzero!(100)));
    })
}

One outcome per record​

Only the unapproved refund comes back, as a copy published after the delay.

examples/kinesis_batches.rs
/// One outcome per refund, in slice order: only the unapproved ones come back.
#[subscriber(KinesisStream::new("refunds"))]
async fn review(batch: &[Refund]) -> Vec<HandlerOutcome> {
    batch
        .iter()
        .map(|refund| {
            if refund.approved {
                HandlerOutcome::ack()
            } else {
                HandlerOutcome::retry_after(RECHECK)
            }
        })
        .collect()
}