Коммиты оффсетов: CommitMessages

Инцидент, который случался почти в каждой команде, работавшей с Kafka: сервис обработки платежей перезапустился после деплоя, и десяток событий исчез из обработки — не зависли в очереди, а просто перестали приходить. Брокер здоров, топик цел, события в логе есть, мониторинг лага показывает ноль. Причина — оффсет группы: позиции читателей живут не в вашем процессе, а на брокере, и если оффсет закоммичен раньше, чем событие обработано, рестарт продолжит после потерянного места. Это не баг kafka-go и не кривые руки — это семантика коммитов, о которой нужно знать до продакшена. В этом уроке разберём, что именно записывает CommitMessages, чем отличается автокоммит kafka от ручного, куда девается ошибка интервального коммита и почему передавать последнее сообщение — легальный способ закоммитить всю пачку. Седьмой урок собрал партиции в группу; здесь мы сделаем так, чтобы группа после рестарта продолжала ровно с того места, где обработка действительно закончилась.

kafka commit offset: что именно записывает kafka-go#

Коммит — это запись на брокере: для пары топик-партиция группа хранит одно число, означающее следующий ожидаемый оффсет. kafka-go вычисляет его из сообщения: смещение сообщения плюс единица. Поэтому коммит сообщения с оффсетом 41 записывает позицию 42 — следующее чтение начнётся с 42. На одну партицию приходится один оффсет, сколько бы сообщений вы ни передали: kafka-go сводит переданные сообщения по партициям и берёт максимум, а документация прямо предупреждает — наибольший оффсет в CommitMessages закоммитит и все предыдущие сообщения партиции. Из этого следует правило последнего сообщения: чтобы двигать партицию вперёд, достаточно передать последнее прочитанное из неё. Отсюда и вторая половина правила: коммит старого сообщения откатывает позицию назад, даже если чтение уже ушло дальше — группы в kafka-go читают и коммитят параллельно, и рассинхрон ловится не брокером, а вами. Коммиты возможны только в групповом режиме: без GroupID метод вернёт ошибку unavailable when GroupID is not set — позиция без группы хранится у вас в коде, как в шестом уроке.

Ручной коммит: FetchMessage плюс CommitMessages

reader := kafka.NewReader(kafka.ReaderConfig{
        Brokers: []string{"localhost:9092"},
        GroupID: "billing",
        Topic:   "payments",
})

for {
        m, err := reader.FetchMessage(ctx) // чтение без коммита
        if err != nil {
                log.Printf("stop: %v", err)
                break
        }

        if err := charge(m); err != nil {
                log.Printf("charge %d failed: %v", m.Offset, err)
        }

        if err := reader.CommitMessages(ctx, m); err != nil {
                log.Printf("commit: %v", err)
                break
        }
}

FetchMessage читает сообщение и ничего не коммитит — в отличие от ReadMessage из шестого урока, который при группе коммитит до возврата. Обработка идёт до коммита, и только успешный её итог подтверждается CommitMessages, записывающим оффсет плюс единица. Ошибка коммита гасит цикл: незакрытая позиция означает, что при рестарте события приедут повторно, и это осознанный выбор в пользу надёжности. Без GroupID метод вернёт ошибку unavailable when GroupID is not set — коммитить без группы нечего.

Автокоммит kafka: ReadMessage и поле CommitInterval#

Автокоммит в kafka-go существует в двух формах, и их путают. Первая — ReadMessage с группой: метод сам коммитит каждое прочитанное сообщение, до возврата из вызова; документация честно оговаривает, что оффсет может закоммититься до завершения обработки. Это удобно и для быстрых обработчиков незаметно, но позиция подтверждается раньше работы — рудимент потери, о котором коллаут ниже. Вторая форма — поле CommitInterval в ReaderConfig: оно не включает автокоммит как магию, а меняет способ доставки ваших же коммитов на брокер. При нуле каждый CommitMessages уходит синхронно: запрос OffsetCommit на брокер, ответ, ошибка возвращается вызывающему. При значении больше нуля коммиты копятся во внутреннем стэше и уходят одной пачкой раз в интервал — меньше запросов при высокой скорости, но ошибка интервального коммита возвращается не вам, а пишется в ErrorLogger внутри Reader. Самокоммит ReadMessage едет через тот же механизм: при нулевом интервале ошибка коммита вернётся из ReadMessage с префиксом committing message, при интервале — осядет в стэше до тикера. Кроме того, стэш коммитов не переживает ребалансировку: при пересборке группы накопленное и не отправленное пропадает. Для тяжёлых потоков это честный компромисс, для редких дорогих событий — лишний риск.

Живое место: потеря при перезапуске на автокоммите

Разбор ночного инцидента: сервис уведомлений падал по OOM, и после каждого рестарта клиенты жаловались на пропавшие письма. Код читал ReadMessage в цикле, обработка занимала секунды — рассылка шла через внешний почтовый API. ReadMessage коммитил оффсет сразу, обработка падала посреди пути, при рестарте группа продолжала после потерянного события. Лечится заменой пары ReadMessage на FetchMessage плюс CommitMessages после успешной отправки — буквально четыре строки диффа. Вывод из истории: автокоммит — не «настройка производительности», а решение о том, какие события можно потерять; принимать его стоит явно, а не дефолтом копипасты. Проверьте свой продюсерский код тем же взглядом: где коммит встречает обработку — там и граница потерь.

Проверка: куда уходит ошибка интервального коммита

ReaderConfig с CommitInterval: time.Second, брокер недоступен, CommitMessages не проходит. Что увидит вызывающий код?

CommitMessages под капотом: синхронный и интервальный пути#

Заглянем в исходник, чтобы путь больше не был магией. При CommitInterval: 0 работает commitLoopImmediate: ваш CommitMessages попадает в канал, воркер сразу отправляет OffsetCommit на брокер с тремя попытками по умолчанию и возвращает результат в ваш канал ошибок — синхронно и с гарантией видимости исхода. При интервале работает commitLoopInterval: тикер раз в CommitInterval снимает накопленный стэш и отправляет одной пачкой; ошибка уходит в ErrorLogger, стэш сбрасывается только при успехе, и — важная деталь из комментария в исходнике — при ребалансе стэш уничтожается, потому что член группы получает новые партиции. Отсюда практическое правило: при частых ребалансах интервальные коммиты теряют накопленное чаще, и ставка на них должна сопровождаться стабильной группой. И помните про offset плюс единица: коммит двигает позицию вперёд, а не «отмечает сообщение прочитанным» — мыслить нужно позицией, а не галочками на событиях. Ошибка unavailable when GroupID is not set возвращается мгновенно и синхронно в обоих режимах: это контракт API, а не сетевой сбой. Увидеть, по какому пути работает ваш Reader, проще всего в логах: при старте группы коммит-цикл пишет started commit в Logger, а дальше поведение при остановленном брокере сразу покажет, синхронный у вас путь или интервальный.

Меню: коммит пачкой по счётчику

const commitEvery = 100

batch := make([]kafka.Message, 0, commitEvery)

for {
        m, err := reader.FetchMessage(ctx)
        if err != nil {
                log.Printf("stop: %v", err)
                break
        }

        process(m)
        batch = append(batch, m)

        if len(batch) >= commitEvery {
                // максимум по партициям: вся пачка уезжает одним вызовом
                if err := reader.CommitMessages(ctx, batch...); err != nil {
                        log.Printf("commit: %v", err)
                }
                batch = batch[:0]
        }
}

if len(batch) > 0 { // хвост без коммита вернётся после рестарта
        if err := reader.CommitMessages(ctx, batch...); err != nil {
                log.Printf("final commit: %v", err)
        }
}

Пачка сообщений передаётся в CommitMessages целиком: kafka-go сам сведёт их по партициям и возьмёт максимум оффсетов по каждой — правило последнего сообщения, которое документация прямо называет легальным способом двигать позицию вперёд. Хвост перед остановкой подтверждается отдельно: иначе после рестарта приедут уже обработанные события. Частота коммитов — ручка между числом запросов к брокеру и размером повтора после падения.

Режимы коммита kafka-go одним столом#

Режим	Как работает	Цена
ReadMessage с группой	Чтение и самокоммит внутри вызова, до вашей обработки	Простая пауза после падения: обработанное и незаконченное теряются
Fetch + Commit, интервал 0	Синхронный OffsetCommit на каждый вызов, ошибка — вызывающему	Больше запросов к брокеру; максимум контроля и минимум потерь
Fetch + Commit, интервал больше 0	Стэш коммитов уходит пачкой раз в CommitInterval	Экономия запросов; ошибки в ErrorLogger, стэш гибнет при ребалансе
Пачка по счётчику	CommitMessages с последним сообщением каждой партиции	Окно повтора в сто сообщений: дубликаты вместо потерь

Проверка: правило последнего сообщения

Партиция одна, прочитаны оффсеты 10 и 11, оба обработаны. Вызов CommitMessages передал только сообщение с оффсетом 10. Что закоммитится?

С какого места продолжит группа?

Прочитайте код и выберите, что он выведет:

// партиция одна; FetchMessage вернул два сообщения:
m0, _ := reader.FetchMessage(ctx) // Offset: 0
m1, _ := reader.FetchMessage(ctx) // Offset: 1

process(m0)
process(m1)

// программист закоммитил только первое:
reader.CommitMessages(ctx, m0)

// процесс убит и перезапущен

Завершение сервиса: коммит хвоста и io.ErrClosedPipe#

Выключение сервиса — второй, наряду с падением, момент, где коммиты решают судьбу событий, и здесь тоже есть порядок. Честная остановка выглядит так: получен сигнал, контекст цикла отменяется — FetchMessage выходит с ошибкой контекста и перестаёт тянуть новые сообщения; дочитанные события довариваются обработчиком; по каждой затронутой партиции уходит финальный CommitMessages с последним обработанным сообщением; и только после этого вызывается Close. Порядок важен: попытка коммитить после закрытия Reader вернёт io.ErrClosedPipe — клиент прямо запрещает коммиты после остановки, потому что член группы без генерации не имеет права двигать позицию. Строка defer reader.Close() при этом не враг: Close на живом Reader штатно останавливает чтение и коммит-цикл, и в простых программах это весь shutdown. Опасно не наличие defer, а последовательность вокруг него: обработка — коммит — Close, и никогда наоборот, иначе хвост обработки окажется незакрытым, а рестарт вернёт ровно те дубликаты, от которых вы строили схему. Поэтому в сервисах с дорогой обработкой shutdown-путь пишется руками: сигнал, отмена, дренирование, финальный коммит, Close — пять шагов, и каждый тестируется так же, как падение.

Одно число на партицию: закоммитить середину нельзя

Kafka хранит на группу ровно один оффсет на топик-партицию, и это ограничение протокола, а не библиотеки: закоммитив позицию 15, вы объявили прочитанным всё до 14 включительно, включая событие с оффсетом 12, на котором упала обработка. Выбрать «закоммитить 15, но оставить 12» невозможно — внутри партиции нет частичных подтверждений. Отсюда устройство ретраев: постоянная ошибка либо блокирует позицию, пока поток стоит, либо требует адресата вроде DLQ из девятнадцатого урока, либо откатывает позицию назад ценой перечитывания. Встроенного «пропустить и идти дальше» в семантике оффсетов нет — и это хорошо: пропуск события должен быть явным решением в коде, а не побочным эффектом коммита.

Оффсеты группы на брокере: retention и переигрывание#

Закоммиченные позиции — тоже данные, и у них есть срок жизни. Поле RetentionTime в ReaderConfig по умолчанию равно минус единице: тогда действует настройка брокера offsets.retention.minutes, которая для Kafka начиная со второй версии составляет семь дней. Группа, не проявлявшая активности неделю, рискует потерять позиции — и после долгого простоя начнёт с StartOffset, по умолчанию FirstOffset, вычитав партиции заново. Отсюда практическое правило для редких потребителей: либо поднимайте retention оффсетов, либо планируйте перечитывание как штатный сценарий. Обратная операция — сдвиг группы назад — из кода группового Reader недоступна: SetOffset с группой возвращает ошибку unavailable when GroupID is set. Переиграть историю группой можно через консольные утилиты сброса оффсетов, новое имя группы или низкоуровневое чтение без группы, о котором отдельный урок про Reader API. Мониторинг лага группы — тема урока про наблюдаемость: закоммиченный оффсет минус конец лога и есть лаг, который вы видите на графиках.

Правила коммитов одним списком#

Коммит — позиция вперёд: оффсет сообщения плюс единица, а не отметка о прочтении.
На партицию один оффсет: максимум по переданным сообщениям, старые закрываются автоматически.
ReadMessage с группой коммитит сам: обработка после коммита — окно потерь.
CommitInterval: 0 — синхронные коммиты с ошибкой в руки; больше нуля — пачка и ErrorLogger.
Стэш интервальных коммитов гибнет при ребалансе: частые пересборки усиливают потери.
Хвост пачки перед остановкой подтверждайте явно — иначе рестарт вернёт обработанное.

Соберите ручной коммит

Допишите чтение без коммита и метод подтверждения позиции.

m, err := reader.(ctx) // чтение БЕЗ коммита if err != nil { break } process(m) if err := reader.(ctx, m); err != nil { log.Printf("commit: %v", err) break }

Соберите цикл с коммитом после обработки

Расставьте строки цикла в порядке исполнения: чтение, выход по ошибке, обработка, коммит, обработка ошибки коммита.

Строки

Ваша программа

Нажимайте на строки слева, чтобы собрать программу

Карточки: коммиты оффсетов

Термин

Коммит оффсета

нажмите, чтобы перевернуть

Определение

Запись позиции группы на брокере: оффсет сообщения плюс единица — следующий ожидаемый. Одна позиция на топик-партицию.

1 / 6

Позиция чтения теперь под контролем: коммит до обработки, после или пачкой — и вы знаете цену каждого варианта, включая выключение сервиса. Из этого урока в код сервиса обычно уезжают две вещи: пара FetchMessage плюс CommitMessages и shutdown-путь с финальным коммитом — остальное решает конфиг. Это ровно тот материал, на котором строится урок про гарантии доставки: at-most-once и at-least-once окажутся не настройками брокера, а расстановкой уже знакомых вызовов. Для чтения без группы и сдвига позиций руками держите в закладках урок про Reader API, а для наблюдения за позициями на живой системе — урок про мониторинг лага. Формат прежний и честный: песочница сеть не поднимает, ручные коммиты тренируются локально против брокера из Docker, интерактивы — программы на чистом Go с той же семантикой.

Песочница урока

Редактируйте код и запускайте его прямо здесь — интерпретатор работает в браузере, ничего устанавливать не нужно. За каждый запуск — +2 XP.

Вывод

Запустите код, чтобы увидеть вывод.

Коротко: что запомнить

Коммит — позиция на брокере: оффсет сообщения плюс единица, одна на топик-партицию.
ReadMessage с группой коммитит сам — до вашей обработки; для медленных обработчиков это окно потерь.
FetchMessage плюс CommitMessages — ручной контроль: обработка до подтверждения, дубликаты вместо потерь.
CommitInterval: 0 — синхронно с ошибкой в руки; больше нуля — пачка, ошибки только в ErrorLogger.
Наибольший оффсет закрывает предыдущие: коммитить достаточно последнее сообщение каждой партиции.
Стэш интервальных коммитов не переживает ребаланс; позиции группы живут offsets.retention минут.
Выключение — тот же контракт, что и падение: обработка, финальный коммит хвоста, Close; коммит после закрытия вернёт io.ErrClosedPipe.
Проверка коммитов — локальный Go и брокер в Docker; песочница показывает семантику на чистом Go.

Домашнее задание

Против брокера из третьего урока соберите группу billing из двух FetchMessage-консьюмеров. Отправьте двадцать платежей, обработайте десять и убейте процесс без финального коммита хвоста: после рестарта убедитесь, что группа продолжила с позиции последнего коммита, и посчитайте дубликаты. Затем замените пары Fetch/Commit на ReadMessage и повторите эксперимент с падением посреди обработки — зафиксируйте потерю. Третий эксперимент: поставьте CommitInterval: time.Second и найдите в логе Reader сообщения ErrorLogger при остановленном брокере, чтобы увидеть, куда уходят ошибки интервальных коммитов. Последний аккорд: отправьте сервису сигнал остановки и убедитесь, что финальный коммит хвоста успел уехать до Close — рестарт после корректного shutdown должен вернуть ноль дубликатов.

Вопросы и ответы по уроку

Как часто коммитить: каждое сообщение или пачкой?

По цене ошибки: события, дубликат которых дороже лишнего запроса, — коммит на каждое сообщение синхронно; потоки, где повтор на сто сообщений ничего не ломает, — пачкой по счётчику или интервалом. Промежуточный разумный старт — коммит каждые 50-100 сообщений или раз в секунду, какой счётчик наступит раньше. И что бы вы ни выбрали, хвост перед остановкой подтверждайте отдельным финальным коммитом: правило последнего сообщения делает это одним вызовом на партицию.

CommitMessages подвисает — это нормально?

Да, при CommitInterval: 0 это синхронный запрос к брокеру: он ждёт ответа или отмены контекста. Если обработка не должна зависеть от брокера, либо укорачивайте контекст коммита, либо переходите на интервальный режим, помня, что его ошибки увидит только ErrorLogger.

Можно ли коммитить без группы?

Нет: CommitMessages без GroupID возвращает ошибку unavailable when GroupID is not set. В режиме чтения партиции позиция — ваша переменная в коде: сохраняйте её сами, если пережить рестарт важно. Коммиты и перечитывание без группы — тема урока про Reader API.

Термины урока

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