Skip to main content

legacy_stream

legacy_stream names the stream a table's bare shard rows belong to, so a table written by an earlier version resumes from them.

Earlier versions of this crate keyed a row by the shard id alone, such as shardId-000000000000. A bare row names no stream, and every stream's first shard has that id. So DynamoLeaseStore reads a bare row only for the stream legacy_stream names. A shard of that stream with no checkpoint under its own stream:shard row resumes from its bare row, and the first read copies the checkpoint to the new row. A checkpoint already on the new row wins over the bare one. Other streams ignore the bare rows. Without legacy_stream, a shard whose bare row holds a checkpoint is refused with a lease error that names the table, the shard and the stream, and it is not read until the stream is named or the row is deleted. Once the upgraded service has opened a subscription on every stream it reads, every bare row with progress has its copy; the rows whose lease_key holds no : can then be deleted, and the setting dropped. A table that served several streams before the upgrade holds, in each bare row, the checkpoint of whichever stream wrote last. No stream can resume from such a row: delete the bare rows and open each stream with start_at at the position it should read from. The two versions lease different rows, so an old and a new instance can read one shard at once; stop the old instances before the new ones start.

Example​

The table returns-leases served the stream returns alone before the upgrade.

examples/kinesis_leases_legacy.rs
/// The table was written by an earlier version for the stream `returns` alone, so its bare rows
/// are that stream's progress.
fn broker(config: &SdkConfig, table: &str) -> KinesisBroker {
    let leases = Arc::new(DynamoLeaseStore::new(config, table).legacy_stream("returns"));
    KinesisBroker::from_config(config.clone()).lease_store(leases)
}