Завершение

Чем ack, drop, retry и retry_after становятся в Kafka, которая не подтверждает записи, а фиксирует одно смещение на партицию.

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

1. Сохранённая позиция: 11 ниже самой ранней выполняемой доставки → 11 готова: сохранена на
2. Сохранённая позиция: 11 ниже самой ранней выполняемой доставки → Зафиксировано смещение 12: с него продолжит группа: автофиксация
3. Одна партиция, смещения в порядке доставки
4. 10 готова
5. 12 идёт
6. 13 готова
7. 14 готова
8. 15 идёт
9. Когда 12 завершится, позиция сдвинется на 14: 15 ещё выполняется

Kafka хранит одно зафиксированное смещение на партицию и группу. Режим Commit подписки решает, как исход обработчика до него доходит.

При Commit::Tracked крейт выключает автоматическое сохранение смещений в librdkafka и запоминает каждое смещение, которое доставляет. Завершение доставки сохраняет позицию сразу под самой ранней доставкой, которая ещё выполняется, а когда таких нет, наибольшее доставленное смещение. Подтверждения из параллельных партиций воркеров приходят в любом порядке, а сохранённая позиция никогда не обгоняет незаконченную работу. Смещения, которых консьюмер не получает, например маркеры транзакций или удалённые при уплотнении записи, её не задерживают. Сохранённую позицию, как обычно, фиксирует таймер librdkafka.

Исход	Commit::Auto (по умолчанию)	Commit::Tracked	Commit::Transactional
ack	Ничего: librdkafka сохранила смещение, когда отдала запись	Смещение завершается, и сохранённая позиция сдвигается вверх	Смещение завершается, конвейер фиксирует его со следующим окном
drop	Ничего, как при ack	Смещение завершается, как при ack	Смещение завершается, как при ack
retry()	Отказ (AckError::Unsupported) и запись в журнал: запись ничто не вернёт	Запись доставляется снова раньше того, что будет выбрано дальше, и позиция её ждёт	Открытое окно не может зафиксироваться: оно прерывается и читается заново

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

При Commit::Transactional консьюмер сам ничего не фиксирует. EosPipeline с тем же идентификатором каждые commit_interval (100 миллисекунд) добавляет завершённые позиции в транзакцию своего продюсера. Поэтому прочитанные смещения и опубликованные записи становятся видны вместе, как показывает страница Транзакции Kafka.

Повторы

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

1. Запись x-ruststream-retry-count говорит, какая это попытка → Обработчик возвращает retry_after(delay) или retry() при объявлении: обработана
2. Обработчик возвращает retry_after(delay) или retry() при объявлении → max_attempts: попытка ниже предела?: повтор
3. max_attempts: попытка ниже предела? → Ниже предела: оригинал завершается, копия уходит в тему: ниже
4. max_attempts: попытка ниже предела? → На пределе: копия уходит в тему dead-letter: на пределе
5. max_attempts: попытка ниже предела? → На пределе, без dead-letter: запись завершается без копии
6. Ниже предела: оригинал завершается, копия уходит в тему → Копия - новая запись в конце своей партиции
7. Отложенную копию держит процесс: падение до срабатывания таймера её теряет

Kafka не придерживает записи и не считает доставки, поэтому и то и другое делает рантайм. Исход retry_after, а также retry() у регистрации, которая объявляет max_attempts или тему dead-letter, идут этим путём в любом режиме фиксации. Номер попытки едет в заголовке копии x-ruststream-retry-count, и первая доставка считается первой попыткой.

Копия уходит по адресу, который сообщает дескриптор. Подписка KafkaTopic сообщает свою тему. Подписка KafkaTopics или KafkaPartitions читает набор, до которого не дотянется ни одна публикация, поэтому её регистрация называет адрес сама, иначе приложение не запустится.

Копия сохраняет ключ записи, поэтому снова попадает в партицию этого ключа, за записями, которые были записаны после. Копию получает каждая группа консьюмеров, которая читает тему, а не только та, что выполнила повтор.