Shards

A subscription lists its stream's shards every ten seconds, leases each one, reads it with one reader, and starts a split's children after their parents.

A Kinesis stream is a set of shards, and the partition key of a record picks its shard: the service hashes the key, so one key always lands on one shard. Records of one shard are read in the order they were written, so records that share a key keep their order. Records of different shards have no order between them, and one subscription reads its shards side by side. For each shard the subscription takes a lease in the lease store; a shard another instance holds is left to that instance. The listing repeats every ten seconds, so a shard that a split or a merge adds is picked up within that time. A child shard starts only after every parent is fully read and checkpointed as finished, which keeps the order of a key across resharding. A parent the subscription never read, while it opens at the tip, does not hold its children back. How a publish names the key is on the Keys page. The in-process transport of a TestApp test keeps one shard per stream, so there every record is ordered against every other.

Example

The stream has two shards, and each customer's shipments carry the customer as their key, so they arrive in order.

Macros
Manual

examples/kinesis_subscriptions_shards.rs
/// Ships each shipment. The stream has two shards, read side by side; a customer's shipments
/// carry the customer as their partition key, so they share a shard and arrive in order.
#[subscriber(KinesisStream::new("shipments").create_if_missing(2))]
async fn ship(shipment: &Shipment) -> HandlerOutcome {
    tracing::info!(shipment.customer, shipment.seq, "shipped");
    HandlerOutcome::ack()
}

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(broker, |b| {
        b.include(ship);
    })
}

examples/manual/kinesis_subscriptions_shards.rs
/// Ships each shipment. The stream has two shards, read side by side; a customer's shipments
/// carry the customer as their partition key, so they share a shard and arrive in order.
struct Ship;

impl Handle<Shipment> for Ship {
    fn handle(
        &self,
        shipment: &Shipment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(shipment.customer, shipment.seq, "shipped");
        ready(Ok(()))
    }
}

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(broker, |b| {
        let shipments = KinesisStream::new("shipments").create_if_missing(2);
        b.include(subscriber(shipments, Ship).build());
    })
}

Example​

Example