How does the Kinesis Client Library distribute a stream's shards across worker instances and remember progress, and what operational problems come from that design?

## Kinesis has no consumer groups, so the client builds one On the service side, Kinesis Data Streams offers shards, iterators, and `GetRecords`/`SubscribeToShard`. There is no server-side notion of a group of workers, no assignment protocol, and no stored progress. Everything about "which of my six machines reads which shard, and where did we get to" is a **client-side** construct, implemented by the Kinesis Client Library (KCL) — and, in the same shape, by the Lambda and Flink integrations. ## The lease table On first start, the KCL creates a DynamoDB table named after the application name you configured. It holds one item per shard: - `leaseKey` — the shard id, and the table's partition key - `leaseOwner` — the worker identifier currently processing it - `leaseCounter` — a monotonically incremented heartbeat - `checkpoint` — the sequence number of the last record the owner declared done (or the sentinels `TRIM_HORIZON`, `LATEST`, `SHARD_END`) - parent shard ids, so resharding can be handled correctly Holding a lease means periodically incrementing `leaseCounter` with a conditional write. A worker takes a lease by conditionally writing its own id when the counter it last observed is unchanged after the failover interval — the classic conditional-write lease. Because the write is conditional on the counter, two workers cannot both win. Balancing rides on the same mechanism: a worker holding fewer than its fair share of leases steals one from the worker holding the most, again by conditional write. Start a second instance of a six-shard application and it converges to three leases each without any coordinator being involved. ## Checkpointing and delivery semantics Your record processor calls `checkpoint()` when it has durably handled records up to a point; the KCL writes that sequence number into the lease item. On failover, the new owner reads the checkpoint and resumes there. That design is squarely **at-least-once**. Records processed but not yet checkpointed when a worker dies are delivered again to whoever picks up the lease. Checkpoint after every record and you minimise duplicates but write to DynamoDB constantly; checkpoint every N records or every few seconds and you trade a wider replay window for far fewer writes. Either way the processing must be idempotent — deduplicate on a business key, or make the write itself idempotent (conditional put, upsert by id). ## Resharding A `SplitShard` or `UpdateShardCount` closes parent shards and opens children. The KCL discovers this, writes lease items for the children recording their parents, and enforces a rule: a child shard is not started until every parent has been read to `SHARD_END` and checkpointed. Since a given partition key lived in the parent before the split and in exactly one child after it, that ordering rule is what preserves per-key ordering across a reshard. It also explains a common observation — after resharding, some new shards sit idle for a while; they are waiting on their parents, not broken. ## The operational problems **The lease table is a dependency you did not ask for.** It is billed, it can be throttled, and it can be deleted by an over-eager cleanup script. Under-provisioned capacity means lease renewals fail, workers believe each other dead, leases churn between them, and throughput collapses even though the stream itself is healthy. Symptoms: rising `GetRecords.IteratorAgeMilliseconds`, DynamoDB throttled-request metrics on that table, and log lines about lost or stolen leases. Provision it for the write rate your checkpoint interval implies, or use on-demand billing for it. **The application name is the identity of progress.** Two deployments (staging and production, blue and green) sharing an application name share one lease table and will each think the other's workers are theirs — shards get processed by the wrong environment. Conversely, renaming the application on a deploy creates a fresh table and the new one starts at whatever initial position you configured. That is either a full reprocess from `TRIM_HORIZON` or a silent gap from `LATEST`, and both have caused real incidents. **Parallelism is capped by shard count.** One lease per shard means at most one worker actively processes a shard. Run twelve instances against a six-shard stream and six sit idle. Scaling the consumer means scaling the stream, which is a slower and more consequential operation than adding pods. **Version differences.** KCL 2.x retrieves via enhanced fan-out by default and can be configured for polling; KCL 1.x polls `GetRecords`. If you migrate and see a new per-consumer-shard-hour line on the bill, that default is why. ## What to monitor Per-shard `GetRecords.IteratorAgeMilliseconds` (the primary lag signal), DynamoDB throttling and consumed capacity on the lease table, and the KCL's own lease-take and lease-loss logs. A healthy application shows stable lease ownership; leases moving constantly between workers is the fingerprint of a lease-table or timeout misconfiguration, not of a slow stream.