Skip to main content

Reading

A consumer reads a shard with an iterator: GetShardIterator places it on the shard's log, and every GetRecords returns records and the next iterator.

The five iterator types placed on a shard's log, and the loop of GetRecords calls that reads from thereOne shard, records in arrival order123456nextTRIM_HORIZONAT_TIMESTAMPAT_SEQUENCE_NUMBERAFTER_SEQUENCE_NUMBERLATESTGetShardIteratorshard, iterator type, positionGetRecordsiterator, limit 1000 or the batch sizeRecords and NextShardIterator:the iterator of the next read;none at the end of a closed shardAn iterator expires 5 minutes after it was issued.An empty read waits poll_interval, 1 s by default
The diagram as text
  1. One shard, records in arrival order: 1; 2; 3; 4; 5; 6; next; TRIM_ HORIZON; AT_ TIMESTAMP; AT_ SEQUENCE_ NUMBER; AFTER_ SEQUENCE_ NUMBER; LATEST
  2. TRIM_ HORIZON → 1
  3. AT_ TIMESTAMP → 2
  4. AT_ SEQUENCE_ NUMBER → 4
  5. AFTER_ SEQUENCE_ NUMBER → 5
  6. LATEST → next
  7. One shard, records in arrival order → GetShardIterator shard, iterator type, position: an iterator type
  8. GetShardIterator shard, iterator type, position → GetRecords iterator, limit 1000 or the batch size: an iterator
  9. GetRecords iterator, limit 1000 or the batch size → Records and NextShardIterator: the iterator of the next read; none at the end of a closed shard
  10. Records and NextShardIterator: the iterator of the next read; none at the end of a closed shard → GetRecords iterator, limit 1000 or the batch size
  11. An iterator expires 5 minutes after it was issued. An empty read waits poll_interval, 1 s by default

The iterator type says where the first read starts. TRIM_HORIZON starts at the oldest record the shard still holds. LATEST starts after the newest one, so the reader sees only what arrives next. AT_TIMESTAMP starts at the first record that arrived at or after a time, by its arrival timestamp. AT_SEQUENCE_NUMBER starts at one record, and AFTER_SEQUENCE_NUMBER right after it.

Each KinesisPosition is one of those types: horizon(), latest(), timestamp(ms) and sequence(shard, seq), which names one record of one shard. A stored checkpoint opens the shard after the record it names. A shard with no checkpoint opens at LATEST, unless start_at(..) names another position. start_at(..) applies at startup, to the shards the subscription opens with, ahead of their checkpoints. A reader started later, for a child shard or for a shard taken over from another instance, resumes from the checkpoint; a child shard with no checkpoint opens at TRIM_HORIZON.

A GetRecords call returns up to 10,000 records or 10 MB. The crate asks for 1,000, or for the batch size of a batch subscription. Every answer carries the next iterator, and the reader uses it for its next call. An answer with no records makes the reader wait the descriptor's poll_interval, one second by default, before it reads again. That keeps the reader inside the five reads a second its shard allows.

An iterator expires five minutes after it was issued. A reader whose iterator expired fetches a new one from its checkpoint, or, before anything is settled, from the first record it read; never from LATEST, so no record is skipped. A throttled read waits one poll interval and tries again with the same iterator. Any other failed read is reported on the subscription's stream, and the reader waits a second and retries its iterator, which is still good. After eleven failures in a row the reader stops and releases its lease; the subscription takes the shard again when it next lists the shards, and the new reader starts where the old one did.

The crate reads with the stream's shared throughput, the five reads a second every consumer of a shard shares. Enhanced fan-out, a dedicated 2 MB a second per consumer over SubscribeToShard, is another read model. A record that a producer aggregated with the Kinesis Producer Library is reported as KinesisError::AggregatedRecord and is not handed to a handler.

The model is Amazon Kinesis Data Streams' own (GetShardIterator, GetRecords, enhanced fan-out, aggregation).