KinesisBatchContext

A batch handler's context, KinesisBatchContext, carries only the seek handle, so asking for a record's position in a batch body does not compile.

A batch spans many records, often from several shards, so it has no single position. The subscription builds the context once per batch from its first record, and it holds what every record of the batch shares: the SeekHandle key, the same key a KinesisContext carries. A batch body takes ctx: &mut Context<'_, KinesisBatchContext> and reads the handle with ctx.context(SeekHandle). A seek from a batch body moves the whole subscription, so the next batch reads from the new position. Ctx<Position> in a batch body is a compile error.

Example

The handler drains jobs two at a time, and a batch that holds the marker job moves the subscription to the tip.

Macros

examples/kinesis_context_batch.rs

/// Drains jobs a batch at a time; a batch holding the marker job moves the subscription to the
/// tip, so the backlog behind it is skipped.
#[subscriber(KinesisStream::new("jobs"))]
async fn drain(jobs: &[Job], ctx: &mut Context<'_, KinesisBatchContext>) -> HandlerOutcome {
    tracing::info!(count = jobs.len(), "jobs drained");
    if jobs.iter().any(|job| job.id == MARKER)
        && ctx
            .context(SeekHandle)
            .seek(KinesisPosition::latest())
            .await
            .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Manual

examples/manual/kinesis_context_batch.rs

/// Drains jobs a batch at a time; a batch holding the marker job moves the subscription to the
/// tip, so the backlog behind it is skipped.
struct Drain;

impl Handle<[Job], (), (), KinesisBatchContext> for Drain {
    async fn handle(
        &self,
        jobs: &[Job],
        _outs: &(),
        ctx: &mut Context<'_, KinesisBatchContext>,
    ) -> Result<(), Vec<HandlerOutcome>> {
        tracing::info!(count = jobs.len(), "jobs drained");
        if !jobs.iter().any(|job| job.id == MARKER) {
            return Ok(());
        }
        ctx.context(SeekHandle)
            .seek(KinesisPosition::latest())
            .await
            .map_err(|_| jobs.iter().map(|_| HandlerOutcome::retry()).collect())
    }
}