Offsets, commit, replay и lag

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

Offset адресует запись в одной партиции

Offset присваивается записи внутри партиции. Он не является идентификатором сущности и не сравнивает время событий. Полный адрес — топик, партиция, offset. Повтор бизнес-события может получить новый адрес, а одинаковое число offset встречается в разных партициях.

В простом журнале есть записи 0…9. Его log end offset, конец журнала, равен 10: это граница после последней записи. Log start offset обозначает нижнюю доступную границу. Если старые сегменты удалены, журнал может начинаться, например, с 4. Между границами иногда встречаются пропуски offsets после compaction или из-за транзакционных записей.

Нумерация существующих записей при очистке сохраняется. Если после compaction остались offsets 4, 8 и 9, читать с позиции 5 означает получить следующую доступную запись, в данном случае 8. Поэтому разность offsets не всегда равна количеству пользовательских событий.

Получено, обработано, сохранено

Position клиента указывает на следующую позицию получения. После poll она может продвинуться сразу для всей полученной пачки, ещё до обработки приложения. Committed offset хранит позицию, с которой группа будет восстанавливаться. Commit здесь означает сохранение позиции чтения, а не запись события в брокер и не commit внешней базы.

В обычном непрерывном журнале после успешной обработки записи 4 нужно сохранить 5. Сохранить 4 означает снова читать запись 4. При пропусках и транзакциях следует использовать следующую позицию, которую предоставляет API клиента для завершённой обработки, а не угадывать её по количеству записей.

Учебная партиция 0 имеет границы [0, 10). Consumer получил записи до 9 включительно, применил до 6, а подтвердил обработку только до 4. Значит, fetch_position = 10, processed_next = 7, committed_offset = 5. После восстановления группа читает с 5. Записи 5 и 6 уже попали в приёмник, поэтому их эффект может повториться.

Для обычных групп Kafka хранит commits во внутреннем топике __consumer_offsets, с привязкой к группе, топику и партиции. Координатор группы обслуживает эти записи. Приложения могут отдельно хранить результаты и checkpoints во внешней системе; это требует своего протокола согласования. Не следует смешивать такой checkpoint с автоматическим commit группы.

Lag измеряет расстояние до выбранной границы

Consumer lag показывает отставание позиции consumer от конца доступного потока. Для административного снимка группы часто считают:

committed_lag = log_end_offset - committed_offset
partition 0: 10 - 5 = 5

Ошибка last_offset - committed_offset дала бы 9 - 5 = 4: последняя запись и граница после неё различаются на единицу. Если же «прочитанный offset» означает последнюю полученную запись 9, сначала нужно перейти к следующей позиции 10.

В учебном непрерывном журнале также видны три разных расстояния: до конца от полученной позиции — 0; получено, но не обработано — 10 - 7 = 3; обработано, но не сохранено — 7 - 5 = 2. Метрика fetch lag клиента и lag по commit могут различаться. Называй измеряемую позицию явно, иначе низкий lag чтения скроет зависшую запись в приёмник.

У read_committed читатель видит транзакционные данные до стабильной границы, LSO, за которой есть незавершённая транзакция. Она может отставать от физического конца журнала. Метрика относительно log end тогда учитывает и записи, ещё недоступные этому читателю. При очистке, транзакциях и дырках в нумерации lag выражает расстояние позиций; для точного количества необработанных фактов нужен отдельный учёт.

Replay начинается с доступной истории

Повторное чтение можно выполнить новой группой с нужной начальной позицией или управляемым изменением сохранённых offsets остановленной группы. Для активной группы произвольный сброс может столкнуться с работающими consumer и их новыми commits. В локальной практике мы сначала остановим читателя, посмотрим план reset и только затем применим его.

auto.offset.reset=earliest задаёт поведение при отсутствии пригодного committed offset: начать с нижней доступной границы. latest начинает у текущего конца. Этот параметр не переписывает существующий действительный commit при каждом старте. Если группа уже сохранила 5, указание earliest само по себе не вернёт её к 0.

Пусть retention удалил offsets меньше 4, а checkpoint равен 2. Восстановить записи 2 и 3 из этого топика уже нельзя. Нужно выбрать политику: остановиться и восстановить историю из другого источника либо явно принять начало с доступной границы. Молчаливый reset может дать неполный отчёт.

Практика

partition_state — синтетический снимок трёх непрерывных журналов без транзакций и compaction. В нём заданы границы, следующая полученная позиция, следующая позиция после завершённой обработки и commit. replay_requests задаёт желаемые позиции повторного чтения. Все величины являются следующими позициями, а не offsets последних обработанных записей.

Посчитай lag, найди работу между получением и commit, затем обнаружь запросы на удалённую историю. SQL проверяет арифметику этого снимка; он не выполняет commit или reset в Kafka.