Consumer подтверждает offset после обработки

Offset в partition, event_id и граница producer ack

Kafka хранит записи в разделённом логе: каждое сообщение имеет offset — порядковый номер внутри конкретной partition. Offset 12 в partition 0 topic telemetry.readings.v1 и offset 12 в partition 1 — это два разных места в разных логах. Consumer group хранит не «последний прочитанный», а следующий: после успешной обработки записи partition 0 / offset 12 зафиксированная позиция равна 13. Именно с неё группа возобновит чтение при рестарте или перебалансировке — при условии что commit прошёл успешно.

event_id — это логический идентификатор публикации, установленный Telemetry в момент записи события. Для нашего учебного примера это event-881, а его запись в Kafka описывается тройкой (telemetry.readings.v1, partition 0, offset 12). Эти два идентификатора дополняют друг друга: offset говорит, где в log находится запись; event_id говорит, что в ней смысловое тождество. Подменять одно другим нельзя — в частности, offset не уникален глобально по topic, а event_id никак не кодирует физическое расположение сообщения в log.

Producer ack, который Telemetry получил в 7.2, означает только одно: брокер принял запись в log с достаточным числом реплик. Он ничего не сообщает о том, прочитал ли consumer group alerts-l73 эту запись, десериализовал ли её, вычислил ли предупреждение и сохранил ли результат. Эти два события — producer ack и consumer offset commit — разделены и независимы по времени. Между ними может пройти сколько угодно: consumer мог ещё не запустить poll, мог уже прочитать запись но ещё не сохранить Alert, а мог упасть на полпути.

ReadingRecorded с event_id=event-881, reading_id=reading-204, device_id=device-17, temperature_celsius=23.4 и observed_at=2026-10-02T09:15:00Z — синтетическая учебная запись, продолжающая трассу 7.1–7.2. Telemetry владеет исходным измерением; Alerts как отдельный consumer создаёт собственный результат и несёт ответственность за его устойчивое сохранение. До тех пор пока Alerts не подтвердил свой результат и не зафиксировал offset, ни один из фактов Telemetry не заменяет эту гарантию.

Синтетический JSON value этой записи выглядит так:

{
  "event_id": "event-881",
  "reading_id": "reading-204",
  "device_id": "device-17",
  "temperature_celsius": 23.4,
  "observed_at": "2026-10-02T09:15:00Z"
}

Franz-go v1.21.0: consumer loop parse → apply → persist → commit

Ниже — учебный consumer для группы alerts-l73. Код оформлен двумя фрагментами: вспомогательные типы и контракты хелперов, затем основной цикл. Фрагменты не скомпилированы с franz-go v1.21.0 и реальным брокером — это аннотированный учебный материал.

Фрагмент A — типы и контракты хелперов (imports для полного Go-файла)

package alertsconsumer

import (
    "context"
    "encoding/json"
    "errors"
    "fmt"

    "github.com/twmb/franz-go/pkg/kgo"
)

// ReadingRecorded — JSON value записи в telemetry.readings.v1.
// Синтетический fixture из 7.1–7.2.
type ReadingRecorded struct {
    EventID     string  `json:"event_id"`
    ReadingID   string  `json:"reading_id"`
    DeviceID    string  `json:"device_id"`
    TempCelsius float64 `json:"temperature_celsius"`
    ObservedAt  string  `json:"observed_at"`
}

// AlertDraft — результат применения доменного правила к событию.
// Конкретный порог здесь не задан; достаточно что apply возвращает
// проект предупреждения, который Store должен сохранить.
type AlertDraft struct {
    EventID  string
    DeviceID string
    Message  string
}

// apply принимает разобранное событие и возвращает проект предупреждения.
// Контракт: чистая функция, без сетевых вызовов, без паники.
func apply(e ReadingRecorded) AlertDraft {
    return AlertDraft{
        EventID:  e.EventID,
        DeviceID: e.DeviceID,
        Message:  fmt.Sprintf("reading %s from %s at %s", e.ReadingID, e.DeviceID, e.ObservedAt),
    }
}

// Store — контракт устойчивого хранилища Alerts.
// Save должен вернуть nil только после подтверждённой записи
// (например, после COMMIT транзакции в PostgreSQL).
// Nil-ответ означает, что результат сохранён; ошибка означает,
// что сохранение не подтверждено — но не обязательно что запись отсутствует,
// если ответ хранилища был потерян.
type Store interface {
    Save(ctx context.Context, draft AlertDraft) error
}

apply — детерминированная функция: одно и то же событие всегда даёт одинаковый AlertDraft. Это важно для трасс отказа: если потребуется повторная обработка той же записи, apply не изменит результат. Реальная дедупликация по event_id — задача 7.4.

Фрагмент B — основной цикл (продолжение того же файла, метод Run)

// Run создаёт consumer, читает по одной записи и следует порядку
// parse → apply → persist → commit. Останавливается при первой ошибке
// на любом шаге. Не возобновляет poll после ошибки.
func Run(ctx context.Context, store Store) error {
    client, err := kgo.NewClient(
        kgo.SeedBrokers("127.0.0.1:9092"),
        kgo.ConsumerGroup("alerts-l73"),
        kgo.ConsumeTopics("telemetry.readings.v1"),
        kgo.DisableAutoCommit(),       // фоновый commit не обгоняет persist
        kgo.BlockRebalanceOnPoll(),    // блокируем перебалансировку между poll и commit
    )
    if err != nil {
        // NewClient проверяет корректность опций, но не устанавливает
        // соединение с брокером в момент вызова.
        return fmt.Errorf("new client: %w", err)
    }
    defer client.Close()

    for {
        // PollRecords(ctx, 1) возвращает не более одной записи суммарно.
        // Это педагогически медленный цикл — не рекомендация для prod.
        fetches := client.PollRecords(ctx, 1)

        // После каждого непустого poll BlockRebalanceOnPoll удерживает
        // rebalance; AllowRebalance должен быть вызван до следующего poll.

        // Отмена контекста — нормальная остановка.
        if errors.Is(ctx.Err(), context.Canceled) {
            client.AllowRebalance()
            return nil
        }

        // Проверяем fetch-ошибки отдельно от записей.
        if errs := fetches.Errors(); len(errs) > 0 {
            client.AllowRebalance()
            return fmt.Errorf("fetch error: %v", errs[0])
        }

        if fetches.Empty() {
            // Пустой poll при отменённом контексте уже обработан выше.
            // Обычный per-poll timeout здесь не задан: используется ctx цикла.
            // Блок начинается после непустого результата; этот вызов
            // AllowRebalance безопасен, но пустой poll сам блок не создаёт.
            client.AllowRebalance()
            continue
        }

        // Берём единственную запись.
        var record *kgo.Record
        fetches.EachRecord(func(r *kgo.Record) { record = r })

        // --- parse ---
        var event ReadingRecorded
        if err := json.Unmarshal(record.Value, &event); err != nil {
            // Невалидный JSON: останавливаемся без commit.
            // Стратегия quarantine — урок 7.5.
            client.AllowRebalance()
            return fmt.Errorf("parse record (partition %d offset %d): %w",
                record.Partition, record.Offset, err)
        }

        // --- apply ---
        draft := apply(event)

        // --- persist ---
        // Save возвращает nil только после подтверждённого commit
        // в хранилище Alerts. Ошибка останавливает цикл без commit offset.
        if err := store.Save(ctx, draft); err != nil {
            client.AllowRebalance()
            return fmt.Errorf("persist alert (event %s): %w", event.EventID, err)
        }

        // --- commit ---
        // CommitRecords синхронно фиксирует позицию record.Offset+1
        // для данной partition в группе alerts-l73.
        if err := client.CommitRecords(ctx, record); err != nil {
            client.AllowRebalance()
            return fmt.Errorf("commit offset (partition %d offset %d): %w",
                record.Partition, record.Offset, err)
        }

        client.AllowRebalance()
    }
}

Несколько вещей важно прочитать именно здесь, а не принять на веру.

DisableAutoCommit — не декоративная опция. Без неё franz-go периодически фиксирует offset в фоне, и этот фоновый commit может опередить store.Save. Тогда при рестарте группа продолжит чтение уже после записи, чей результат никогда не попал в хранилище Alerts.

CommitRecords(ctx, record) фиксирует позицию record.Offset + 1 для конкретной partition — то есть 13 для нашего partition 0 / offset 12. Это именно тот синхронный вызов, который возвращает ошибку при сетевом сбое или потере ответа брокера. Проверка ошибки обязательна.

store.Save контрактно возвращает nil только после подтверждённой записи. Если реализация вернула ошибку, мы не знаем наверняка, записан ли результат — ответ хранилища мог потеряться. Именно поэтому ошибка store.Save формулируется как «сохранение не подтверждено», а не как «запись точно отсутствует».

Rebalance: BlockRebalanceOnPoll, AllowRebalance и когда цикл должен остановиться

BlockRebalanceOnPoll — опция franz-go, которая после каждого непустого PollRecords удерживает перебалансировку partition до явного вызова AllowRebalance. Смысл: пока consumer обрабатывает запись и не дошёл до CommitRecords, брокер не переназначит partition другому участнику группы в рамках той же группы. Это сужает окно, в котором другой участник той же группы alerts-l73 мог бы начать читать с позиции, которую мы ещё не подтвердили.

Но это не абсолютная защита. Heartbeat поддерживается клиентом в фоне, поэтому длительность обработки сама по себе не означает превышение session timeout. Если rebalance уже начался, задержка освобождения блока может превысить rebalance timeout и привести к исключению участника из группы. BlockRebalanceOnPoll не делает участника бессмертным; он лишь откладывает согласование rebalance на стороне клиента до точки AllowRebalance.

Правило: AllowRebalance должен быть вызван на каждом выходе из тела цикла, включая выходы по ошибке. В коде выше это выполнено явно на каждой ветви return. Нельзя оборачивать AllowRebalance в defer внутри функции с неограниченным циклом — defer сработает только при выходе из функции, а не на каждой итерации цикла. Между двумя вызовами PollRecords rebalance должен быть разблокирован.

Вот антипример, который нарушает это правило:

// НЕПРАВИЛЬНО: defer не срабатывает на каждой итерации цикла.
for {
    fetches := client.PollRecords(ctx, 1)
    defer client.AllowRebalance() // сработает только при выходе из функции Run

    if err := store.Save(ctx, draft); err != nil {
        return err // здесь defer сработает при выходе из Run
    }
    // ...
}

При успешных итерациях антипримера deferred-вызовы накапливаются и не освобождают блок перед следующим poll. При return они выполнятся, но следующего poll уже нет. В правильном варианте из раздела 2 AllowRebalance вызывается после обработки каждого непустого результата, в том числе перед выходом по ошибке; безопасный вызов на пустом результате не означает, что он создал блок.

Остановка цикла при ошибке — не пессимизм, а необходимость. Если json.Unmarshal вернул ошибку, групповая позиция для этой записи не подтверждена. Но текущая fetch-позиция consumer в памяти уже продвинулась: следующий poll того же процесса может вернуть более поздние записи, а не снова offset 12. Повтор offset 12 возможен после рестарта или переназначения partition при возобновлении с сохранённой групповой позиции. Правильная реакция на ошибке parse — выйти из цикла и передать решение наверх; что делать с невалидной записью (пропустить, поместить в карантин, алертить) — отдельная политика, которую рассмотрит урок 7.5.

Аналогично при ошибке store.Save или CommitRecords: продолжать poll — значит рисковать продвинуться вперёд по partition без подтверждённого результата для текущей записи. Правило цикла в этом уроке однозначное: любая ошибка на шагах parse / apply / persist / commit останавливает цикл.

Разграничение между отменой контекста и реальной fetch-ошибкой также принципиально. context.Canceled — ожидаемый сигнал штатного завершения; его нужно обработать как нормальный выход, не как сбой. Любая другая ошибка в fetches.Errors() — повод остановить цикл и залогировать причину.

Трассы отказа: что известно и что неизвестно в каждой точке краша

Рассмотрим partition 0, запись с offset 12, event_id=event-881. После успешного CommitRecords зафиксированная позиция группы alerts-l73 в этой partition равна 13.

Условная трасса для этой же записи:

Статичная схема (PNG)

partition 0 / offset 12: poll → persist подтверждён → crash до commit(13)
restart той же группы: offset 12 может быть прочитан повторно

Краш до завершения store.Save

Нужно различать две точки остановки. Если consumer упал точно до устойчивой записи хранилищем, нового результата этой попытки нет. Если он лишь не дождался nil от store.Save, сохранение не подтверждено вызывающему, но могло уже завершиться до краша или потери ответа. В обоих случаях offset этой попыткой не зафиксирован; при возобновлении с групповой позиции запись offset 12 может прийти повторно. Во втором случае повтор потенциально создаёт дубликат, поэтому отсутствие ответа не делает его автоматически безопасным.

Краш после store.Save вернул nil, до CommitRecords

Это ключевое окно. Результат Alerts уже записан в хранилище — store.Save вернул nil, что по контракту означает подтверждённый commit транзакции. Но consumer упал до вызова CommitRecords. Kafka не знает, что эта группа обработала offset 12. При рестарте группа может повторно прочитать запись и вызвать store.Save для того же event_id=event-881. Без защиты повтор может создать дубликат; фактический эффект зависит от ограничений и логики хранилища Alerts. Именно по этой причине 7.4 вводит маркер и транзакцию: commit результата и коммит маркера «эта запись обработана» выполняются атомарно на стороне Alerts DB, прежде чем подтверждается offset.

Краш после CommitRecords вернул nil

Результат сохранён, позиция 13 зафиксирована в брокере. При обычном возобновлении той же группы без ручного сброса offsets она продолжит с offset 13, если зафиксированная позиция доступна. В этом сценарии запись offset 12 пропускается при чтении; это не запрет на отдельное ручное перечитывание. Это наиболее «спокойный» сценарий, но он всё равно не означает атомарности между Alerts DB и Kafka: это два отдельных подтверждения, просто оба успели завершиться до краша.

Ответ на commit-запрос потерян

store.Save вернул nil — результат сохранён. CommitRecords отправил запрос брокеру, но ответ не пришёл: сеть упала, клиент завис, процесс убили. Вызывающий код видит ошибку или вообще не видит ответа. Что неизвестно: дошёл ли запрос до брокера и был ли commit выполнен. Нельзя утверждать ни что «Kafka точно не зафиксировала позицию», ни что «Kafka точно зафиксировала». При рестарте запись может прийти снова — или нет, в зависимости от реального состояния брокера. Идемпотентность на стороне Alerts позволяет безопасно обработать возможный повтор, но сама по себе не раскрывает, был ли offset commit выполнен брокером после потерянного ответа.

Ошибочный порядок: commit до persist

Для контраста — что происходит, если сначала вызвать CommitRecords, а потом store.Save:

// НЕПРАВИЛЬНО: offset зафиксирован до устойчивой записи результата.
if err := client.CommitRecords(ctx, record); err != nil {
    client.AllowRebalance()
    return err
}
// Краш здесь → обычное возобновление группы пропустит offset 12.
if err := store.Save(ctx, draft); err != nil {
    client.AllowRebalance()
    return err
}

Если commit подтвердился, а процесс упал до вызова store.Save, при обычном возобновлении с доступной позицией группа продолжит с offset 13. Эта попытка не создала результат Alerts, а обычное чтение пропустит offset 12. Ручной сброс offsets или отдельное переобрабатывание могут вернуть запись в работу; слово «навсегда» здесь неверно. В отличие от окна после persist до commit, где возможен дубль, commit-first создаёт риск пропуска необходимого эффекта.

Из четырёх сценариев только один даёт уверенное «всё в порядке» — и только если оба подтверждения (store и commit) успели завершиться до краша. В остальных случаях результат может отсутствовать при доказанном краше до записи, оставаться неподтверждённым до ответа store или быть уже сохранённым с возможным повтором до commit. Потерянный ответ commit оставляет неизвестной групповую позицию. Порядок persist → commit сохраняет возможность повторной обработки; защита по event_id в 7.4 ограничит повторный эффект. Пропуск после commit-first требует отдельного обнаружения и переобработки, а не обычного возобновления группы.