Чтение

Потребитель читает сегмент итератором: GetShardIterator ставит его на журнал сегмента, а каждый GetRecords возвращает записи и следующий итератор.

Схема текстом

1. Один сегмент, записи в порядке поступления: 1; 2; 3; 4; 5; 6; следующая; 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 → следующая
7. Один сегмент, записи в порядке поступления → GetShardIterator сегмент, тип итератора, позиция: тип итератора
8. GetShardIterator сегмент, тип итератора, позиция → GetRecords итератор, лимит 1000 или размер пакета: итератор
9. GetRecords итератор, лимит 1000 или размер пакета → Records и NextShardIterator: итератор следующего чтения; в конце закрытого сегмента его нет
10. Records и NextShardIterator: итератор следующего чтения; в конце закрытого сегмента его нет → GetRecords итератор, лимит 1000 или размер пакета
11. Итератор истекает через 5 минут после выдачи. После пустого чтения - пауза poll_interval, по умолчанию 1 с

Тип итератора определяет, где начинается первое чтение. TRIM_HORIZON начинает с самой старой записи, которую сегмент ещё хранит. LATEST начинает после самой новой, поэтому читатель видит только то, что поступит дальше. AT_TIMESTAMP начинает с первой записи, поступившей в заданное время или позже, по метке времени поступления. AT_SEQUENCE_NUMBER начинает с одной записи, а AFTER_SEQUENCE_NUMBER - сразу после неё.

Каждая позиция KinesisPosition - один из этих типов: horizon(), latest(), timestamp(ms) и sequence(shard, seq), которая указывает на одну запись одного сегмента. Сохранённая контрольная точка открывает сегмент после записи, на которую она указывает. Сегмент без контрольной точки открывается с LATEST, если start_at(..) не задаёт другую позицию. start_at(..) действует при старте, на сегменты, с которыми открывается подписка, и имеет приоритет над их контрольными точками. Читатель, запущенный позже, для дочернего сегмента или для сегмента, перешедшего от другого экземпляра, продолжает с контрольной точки. Дочерний сегмент без контрольной точки открывается с TRIM_HORIZON.

Вызов GetRecords возвращает до 10 000 записей или 10 МБ. Крейт запрашивает 1000 записей, а у подписки на пакеты - размер пакета. Каждый ответ содержит следующий итератор, и читатель использует его в следующем вызове. После ответа без записей читатель ждёт poll_interval дескриптора, по умолчанию одну секунду, и только потом читает снова. Так читатель укладывается в пять чтений в секунду, которые разрешает его сегмент.

Итератор истекает через пять минут после выдачи. Читатель, у которого истёк итератор, получает новый от своей контрольной точки, а пока ничего не завершено, от первой прочитанной записи, но никогда от LATEST, поэтому ни одна запись не пропускается. Чтение, ограниченное по частоте, ждёт один интервал опроса и повторяется с тем же итератором. Любую другую ошибку чтения читатель сообщает в потоке доставок подписки, ждёт секунду и повторяет чтение с тем же итератором: он ещё действует. После одиннадцати ошибок подряд читатель останавливается и освобождает аренду. Подписка снова берёт сегмент, когда в следующий раз перечисляет сегменты, и новый читатель начинает там, где начал прежний.

Крейт читает с общей пропускной способностью потока: пять чтений в секунду, которые делят все потребители сегмента. Расширенное разветвление, выделенные 2 МБ в секунду на потребителя через SubscribeToShard, - другая модель чтения. Запись, которую производитель агрегировал с помощью Kinesis Producer Library, сообщается как ошибка KinesisError::AggregatedRecord и не передаётся обработчику.

Эта модель взята у самого Amazon Kinesis Data Streams (GetShardIterator, GetRecords, расширенное разветвление, агрегация).