8.2Kafka

Представь Kafka как упорядоченный лог с независимыми offset, а не как очередь, где сообщение исчезает после чтения. Свяжи partition key с порядком, consumer group с распределением работы и commit offset с повторной доставкой.

Что такое лог и чем он не очередь

Большинство недоразумений вокруг Kafka растёт из одного места: человек читает «Kafka - брокер сообщений» и достраивает привычную картинку «положил, забрал, исчезло». В Kafka сообщение никуда не исчезает.

1. Лог - это файл, в который только дописывают

Слово лог здесь означает не «логи приложения», а структуру данных: последовательность записей, к которой можно только дописать в конец. Изменить или удалить запись в середине нельзя - такой операции просто нет. По-английски это называется append-only log, «журнал только на дозапись».

Так устроен бумажный кассовый журнал. Каждая операция пишется новой строкой, строки нумеруются подряд, вымарывать написанное запрещено. Ошибся - пишешь новую строку «отмена операции №17». Такой журнал можно перечитать с начала и восстановить всю историю: из него ничего не пропало.

2. Читатель - это закладка, а не потребитель

В очереди чтение сообщение расходует: забрал - его больше нет. В логе чтение не меняет ничего. Каждый читатель держит собственный номер строки, на которой он остановился; этот номер и называется offset («смещение»). Прочитать запись означает взять строку под своим номером и сдвинуть номер на единицу вперёд.

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

3. Чем это отличается от очереди - практически

В эксплуатации это значит, что Kafka не знает, «сколько сообщений в очереди». Она знает, где конец лога и где чья закладка. Разница между ними называется lag, и это единственная метрика, которая тут вообще имеет смысл.

Архитектура: из чего это собрано

Кластер состоит из нескольких брокеров (процессов JVM), у каждого свой набор дисков. Топик остаётся логическим именем, физического воплощения у него нет. Хранятся данные в партициях: партиция занимает каталог на диске конкретного брокера, внутри лежит append-only лог. Топик с 12 партициями даёт 12 независимых логов, размазанных по кластеру.

Партиция нарезана на сегменты, каждый из трёх файлов:

• 00000000000000012345.log хранит сами записи, батчами, ровно в том виде, в каком их прислал продюсер;
• .index - разреженный индекс «относительный offset → байтовое смещение в .log» (запись примерно раз в index.interval.bytes = 4 КБ);
• .timeindex переводит «timestamp → offset» и нужен для retention.ms и для offsetsForTimes() при перемотке по времени.

Число в имени файла, base offset, совпадает с offset первой записи сегмента. Пишут всегда только в активный сегмент; он закрывается и создаётся новый по segment.bytes (1 ГБ по умолчанию) или segment.ms (7 суток). Удаление и компакция работают сегментами, а не записями, и активный сегмент не трогают никогда. При низком трафике «протухшее» сообщение может пролежать сильно дольше retention.ms.

Offset служит номером записи внутри партиции: номера монотонно растут, назначает их лидер. Он не глобальный: offset 100 в P0 и offset 100 в P1 не связаны ничем. Запись по offset брокер ищет так: бинарным поиском по именам сегментов, затем по .index, затем линейным досканированием внутри 4 КБ. Выходит, что чтение с произвольного offset стоит примерно столько же, сколько чтение с конца.

У каждой партиции есть лидер и фолловеры (replication.factor). Весь трафик записи и (по умолчанию) чтения идёт через лидера; фолловеры непрерывно тянут данные тем же fetch-протоколом, что и обычные консьюмеры. В ISR входят реплики, которые не отстали больше чем на replica.lag.time.max.ms (30 с). High watermark отмечает максимальный offset, реплицированный на все реплики из ISR; консьюмеры видят записи только до HW, иначе после смены лидера они могли бы прочитать то, чего в новом лидере нет.

Координацию (кто лидер, где партиции, кто в ISR) раньше вёл ZooKeeper. После KIP-500 метаданные живут в самой Kafka, в режиме KRaft: выделенные controller-узлы держат Raft-кворум и ведут собственный лог метаданных. Production-ready с 3.3, миграция с 3.6, в Kafka 4.0 (2025) ZooKeeper удалён полностью. На практике failover контроллера стал на порядок быстрее, а кластер держит миллионы партиций вместо пары сотен тысяч.

Почему Kafka быстрая

Вопрос любят задавать, потому что ответ проверяет понимание работы с диском и сетью, а не знание Kafka. Четыре причины, по убыванию значимости.

1. Последовательная запись append-only. Нет обновлений на месте, нет B-дерева, нет случайных seek. Последовательная запись на обычный HDD даёт сотни МБ/с, что сопоставимо со случайным доступом к памяти. Kafka нарочно спроектирована так, чтобы обращаться к диску только последовательно.
2. Page cache вместо своего кэша. Kafka не держит данные в JVM heap, а пишет и читает через страничный кэш ядра. GC от этого не страдает, кэш переживает перезапуск брокера, а «горячие» читатели (те, кто у хвоста лога) вообще не касаются диска. Отсюда практическое правило: брокеру дают маленький heap (6–8 ГБ), а всю остальную память отдают ОС.
3. Zero-copy. Консьюмеру данные отдаются через sendfile(2): page cache → сокет, без копирования в user space и обратно. Это экономит два копирования и два переключения контекста. Оговорка: zero-copy отваливается, если включён TLS (данные надо шифровать в user space) или если брокеру приходится конвертировать формат записи под старого клиента. Отсюда и ответ на вопрос «почему включение TLS уронило пропускную способность вдвое».
4. Батчинг и единый формат. Продюсер собирает record batch, сжимает его целиком (а не каждое сообщение), брокер кладёт батч на диск как есть и как есть отдаёт консьюмеру. Распаковки и перепаковки на брокере нет, и стоимость сетевого и дискового вызова размазывается на тысячи сообщений.

Producer: путь одной записи

Send() в Kafka-клиенте не отправляет ничего. Он сериализует ключ и значение, спрашивает у partitioner номер партиции, кладёт запись в аккумулятор (буфер батчей, по батчу на партицию) и возвращает управление. Отдельный поток-отправитель забирает готовые батчи, группирует их по брокерам-лидерам и шлёт один запрос на брокер.

acks и min.insync.replicas

Идемпотентный продюсер

Продюсер отправил батч, брокер записал, ответ потерялся в сети, продюсер повторил, и в логе две одинаковые записи. Идемпотентность (enable.idempotence=true, включена по умолчанию с Kafka 3.0) это лечит: продюсер при старте получает PID и ведёт sequence number отдельно на каждую партицию. Брокер помнит последний принятый номер по паре (PID, партиция): пришёл тот же номер - DuplicateSequenceNumber, батч молча отбрасывается с успешным ответом; пришёл номер с разрывом - OutOfOrderSequenceException.

Побочный эффект, за который её и любят: идемпотентность чинит порядок при ретраях. Без неё при max.in.flight.requests.per.connection > 1 повторно отправленный первый батч приедет после второго. С ней брокер по sequence number отклоняет батч с разрывом в номерах, продюсер отправляет заново по порядку, и до 5 in-flight безопасно. Границы гарантии: одна сессия продюсера, одна партиция. Перезапустили процесс - новый PID, дедупликация не работает. На другую партицию она тоже не распространяется.

Ключ, партиция и порядок

// Ключ есть   -> partition = murmur2(key) % numPartitions  (детерминированно)
// Ключа нет   -> Java-клиент: sticky partitioner, льёт в одну партицию, пока не закроется
//                батч, и переключается на следующую. kafka-go Murmur2Balancer берёт
//                для записи без ключа случайную партицию.
w := &kafka.Writer{
    Addr:     kafka.TCP("kafka-1:9092", "kafka-2:9092"),
    Topic:    "orders",
    Balancer: &kafka.Murmur2Balancer{}, // как в Java-клиенте. kafka.Hash{} считает FNV-1a
                                      // и совпадает с Sarama, но не с Java
    RequiredAcks: kafka.RequireAll,   // acks=all
    Async:    false,                  // Async=true = потеря при падении процесса
    BatchTimeout: 10 * time.Millisecond, // аналог linger.ms
    BatchSize:    100,
    Compression:  kafka.Lz4,
}
err := w.WriteMessages(ctx, kafka.Message{
    Key:   []byte(orderID),           // всё про один заказ -> одна партиция -> порядок
    Value: payload,
    Headers: []kafka.Header{
        {Key: "event_id",    Value: []byte(eventID)},   // для дедупликации у потребителя
        {Key: "traceparent", Value: []byte(traceID)},   // трейсинг сквозь брокер
    },
})

Батчинг, linger.ms и компрессия

batch.size (16 КБ) ограничивает батч на партицию по байтам. linger.ms (с Kafka 4.0 по умолчанию 5 мс, раньше 0) говорит, сколько ждать доукомплектования батча перед отправкой. Даже ноль не означает «без батчинга»: батч всё равно собирается за время, пока предыдущий запрос в полёте. Дешевле всего поднять пропускную способность в разы, увеличив linger.ms до 5–20 мс: батчи крупнее, компрессия эффективнее, запросов меньше. Платим фиксированной добавкой к задержке.

Сжимается батч целиком, и чем он крупнее, тем выгоднее компрессия. На практике lz4 почти всегда правильный дефолт (быстрый, приличная степень); zstd (с Kafka 2.1) жмёт заметно лучше при чуть большем CPU и хорош для JSON и дорогого трафика между зонами; gzip самый медленный, брать его не стоит; snappy остался легаси-компромиссом. Сжатый батч так и лежит на диске и в том же виде уходит консьюмеру: брокер его не распаковывает (кроме случаев, когда включена валидация или конвертация формата).

Consumer groups и ребалансировка

Consumer group объединяет несколько процессов, которые вместе читают один топик и делят его партиции: каждая партиция достаётся ровно одному участнику группы. Группа распараллеливает обработку, не ломая порядок внутри партиции. Ребалансировкой называют перераздачу партиций между участниками: кто-то пришёл, кто-то ушёл или перестал отвечать, и надо заново решить, кто какие партиции читает. Похоже на официантов, которые делят столы в зале: вышел один покурить, его столы срочно раздают остальным, и на время передела обслуживание встаёт.

Консьюмеры объединяются в группу по group.id. Один брокер становится для этой группы group coordinator (выбирается по хешу group.id): ведёт список членов, раздаёт партиции и хранит offset. А вот считает раздачу лидер группы, один из консьюмеров: координатор присылает ему список всех членов и подписок, тот раскладывает партиции по выбранной стратегии и возвращает план. Логика назначения остаётся в клиенте, и её можно менять без обновления кластера.

Ребалансировка запускается, когда меняется состав группы или набор партиций:

• новый консьюмер вызвал Subscribe и прислал JoinGroup;
• консьюмер вышел штатно (Close шлёт LeaveGroup);
• консьюмер перестал слать heartbeat дольше session.timeout.ms (45 с в новых версиях, heartbeat идёт из отдельного потока каждые heartbeat.interval.ms = 3 с);
• консьюмер не вызвал poll() дольше max.poll.interval.ms (5 минут) - такой считается зависшим, даже если heartbeat идёт;
• в топик добавили партиции или изменилась подписка по регулярке.

Стратегии назначения

Коммит offset

Committed offset равен offset следующей записи, которую надо прочитать, то есть «последняя обработанная + 1». Хранится он в служебном топике __consumer_offsets (50 партиций, cleanup.policy=compact) под ключом (group.id, topic, partition). Компакция нужна, чтобы топик не рос бесконечно: по каждому ключу важно только последнее значение.

Авто-коммит (enable.auto.commit=true, интервал 5 с) коммитит не по таймеру в фоне, а внутри вызова poll(): если с прошлого коммита прошло больше интервала, коммитится всё, что было выдано предыдущим poll. Пока цикл синхронный, эти записи к тому моменту уже обработаны, и выходит тот же at-least-once. Опасно, когда обработку отдают в другие горутины: тогда коммит подтверждает записи, обработку которых ты ещё не завершил, и падение в этом окне теряет сообщение навсегда. В librdkafka и в kafka-go с ReadMessage offset запоминается уже при выдаче, так что коммит может обогнать обработку и без всяких горутин.

// segmentio/kafka-go: ручной коммит после успешной обработки.
r := kafka.NewReader(kafka.ReaderConfig{
    Brokers:  []string{"kafka-1:9092"},
    GroupID:  "billing",              // GroupID включает управление группой
    Topic:    "orders",
    MinBytes: 10e3,                   // fetch.min.bytes
    MaxBytes: 10e6,
    MaxWait:  500 * time.Millisecond, // fetch.max.wait.ms
    // CommitInterval = 0: CommitMessages коммитит синхронно
})
defer r.Close()

for {
    // FetchMessage, а не ReadMessage: ReadMessage коммитит сам
    m, err := r.FetchMessage(ctx)
    if err != nil {
        return err                    // ctx отменён или reader закрыт
    }
    if err := handle(ctx, m); err != nil {
        // continue здесь потерял бы сообщение: коммит следующего сдвинет
        // offset и за него. Откладываем в retry-топик, потом коммитим.
        if err := toRetry(ctx, m); err != nil {
            return err                // не отложили - выходим без коммита
        }
    }
    if err := r.CommitMessages(ctx, m); err != nil {
        slog.Error("commit", "err", err)
    }
}

Retention и log compaction

Retention определяет, сколько времени или байт Kafka держит записи, прежде чем выбросить. Log compaction убирает иначе: вместо «выбросить старое по времени» брокер выбрасывает устаревшие версии, оставляя по каждому ключу только последнюю запись. Аналогия из магазина: retention говорит «храним чеки за три месяца, дальше в мусор», а compaction - «по каждому товару держим только актуальный ценник, старые выбрасываем, сколько бы им ни было лет».

Общее для обеих политик: Kafka удаляет данные по cleanup.policy, а не по факту прочтения. Никто не прочитал - всё равно удалит по сроку; все прочитали - всё равно будет хранить до срока.

• delete (дефолт): целые сегменты удаляются, когда все их записи старше retention.ms (7 суток) либо когда суммарный размер партиции превысил retention.bytes. Обе границы работают одновременно: что наступит раньше. Считать надо на партицию, а не на топик: retention.bytes=1GB при 24 партициях дают 24 ГБ на топик, и ещё умножить на replication factor.
• compact - размер ограничен не временем, а числом уникальных ключей: для каждого ключа хранится последнее значение. Топик превращается из журнала изменений в материализованный снимок состояния, который можно перечитать с нуля и восстановить всю таблицу.
• compact,delete включает обе: последнее значение по ключу, но не старше retention. Так делают, например, «состояние за последние 30 дней».

Consumer lag

Lag показывает, сколько записей в партиции уже написано, но этой группой ещё не обработано. Формально lag = log end offset − committed offset для каждой пары (группа, партиция): конец лога минус закладка. Это как стопка непрочитанных писем в ящике: важно не то, что там что-то лежит, а растёт она или тает.

По лагу судят о здоровье асинхронной системы: он показывает не «сколько сообщений в очереди», а насколько потребитель отстал от реальности.

Чем мерить

• kafka-consumer-groups.sh --describe --group billing - быстро и руками, показывает лаг по каждой партиции.
• kafka-exporter отдаёт в Prometheus метрику kafka_consumergroup_lag с разбивкой по партициям (у kafka-lag-exporter она называется kafka_consumergroup_group_lag). Алертить надо на производную (лаг растёт N минут подряд), а не на абсолютное значение: 50 000 при потоке 100k/с дают всего полсекунды отставания, и это нормально.
• Lag в единицах времени лучше всего подходит для SLO: «отстаём на 8 секунд» понятно бизнесу, а «отстаём на 400 000 сообщений» нет. Считается по .timeindex или разностью timestamp последней обработанной записи и текущего времени.
• Burrow (LinkedIn) смотрит не на сам лаг, а на его тренд, и выдаёт статус OK/WARN/ERR без ручных порогов.

Лаг растёт: что смотреть по порядку

Ретраи и DLQ

У consumer group в Kafka нет ни встроенного requeue, ни встроенного DLQ. Нельзя «вернуть сообщение в очередь»: указатель всего один, offset, и он либо двигается, либо нет. Вернуть отдельную запись умеет только share group из Kafka 4.2 (см. выше). Остаются два примитивных подхода, и оба плохи:

• Не коммитить и читать заново. Партиция встаёт колом на одном ядовитом сообщении (head-of-line blocking): за ним копятся тысячи корректных, а мы в бесконечном цикле падаем на одном. Вдобавок так быстро превышается max.poll.interval.ms, и начинается ребалансировка.
• Ретраить в цикле внутри обработчика. То же самое: пока спим между попытками, не вызываем poll(). Работает только при коротком общем бюджете (несколько секунд) и обязательно с pause() партиции и дальнейшими вызовами poll().

В проде делают цепочку retry-топиков с возрастающей задержкой и финальный DLQ.

# Топология
orders                 -> основной консьюмер
orders.retry.5s        -> консьюмер, который перед обработкой спит до (produced_at + 5s)
orders.retry.1m
orders.retry.10m
orders.DLQ             -> никто не читает автоматически; читают глазами и руками

# Заголовки, которые надо переносить вместе с сообщением:
x-original-topic       # откуда пришло изначально
x-retry-count          # сколько попыток уже было
x-first-failed-at      # когда сломалось впервые
x-last-error           # тип и текст последней ошибки
x-original-offset      # координаты оригинала, чтобы найти его в логе

// Консьюмер retry-топика спит до времени готовности сообщения.
// У всех сообщений в orders.retry.1m одинаковая задержка,
// поэтому топик упорядочен по времени готовности и достаточно
// подождать голову: приоритетный разбор не нужен.
func (c *RetryConsumer) run(ctx context.Context) error {
    for {
        m, err := c.reader.FetchMessage(ctx)
        if err != nil {
            return err
        }
        ready := m.Time.Add(c.delay)
        if d := time.Until(ready); d > 0 {
            // kafka-go держит членство в группе фоновым heartbeat, спать можно.
            // Java-клиенту на это время нужны pause() и дальнейшие вызовы poll().
            select {
            case <-time.After(d):
            case <-ctx.Done():
                return ctx.Err()
            }
        }
        var perr error
        switch err := c.handle(ctx, m); {
        case err == nil:
            // успех
        case errors.Is(err, ErrPermanent) || retryCount(m) >= c.maxRetries:
            perr = c.produce(ctx, c.dlqTopic, m) // в DLQ без дальнейших попыток
        default:
            perr = c.produce(ctx, c.nextTopic, m) // на следующую ступень задержки
        }
        if perr != nil {
            return perr // не переложили - не коммитим, прочитаем снова
        }
        if err := c.reader.CommitMessages(ctx, m); err != nil {
            return err
        }
    }
}

Транзакции и exactly-once

Транзакции Kafka закрывают ровно один сценарий: read-process-write внутри Kafka. Прочитали из топика A, посчитали, записали в топики B и C, и всё это либо видно потребителям целиком, либо не видно вовсе.

Механика по шагам:

1. Продюсер задаёт transactional.id, стабильный идентификатор, переживающий рестарт (например, billing-worker-p3, привязанный к партиции, а не случайный UUID).
2. initTransactions() находит transaction coordinator и получает PID с новым epoch. Все старые продюсеры с тем же transactional.id и меньшим epoch получают ProducerFenced. Так работает zombie fencing: подвисший старый инстанс не сможет дописать в транзакцию после того, как его заменили.
3. Записи внутри транзакции пишутся в целевые партиции сразу, но помечены как транзакционные.
4. sendOffsetsToTransaction(offsets, consumer.groupMetadata()) кладёт коммит входных offset в ту же транзакцию, и связка «прочитал и записал» становится атомарной.
5. commitTransaction(): координатор пишет решение в __transaction_state и рассылает во все затронутые партиции control batch (маркер COMMIT или ABORT).

На стороне чтения нужен isolation.level=read_committed. Такой консьюмер видит записи только до LSO и отфильтровывает записи прерванных транзакций по маркерам. А по умолчанию стоит read_uncommitted, отсюда классический подвох «включили транзакции, а дубликаты и грязные данные остались»: консьюмера просто забыли переключить.

Библиотеки Go

Что стоит запомнить

• Kafka хранит записи до окончания retention; чтение не удаляет их.
• Порядок гарантируется внутри partition, а одинаковый key обычно направляет связанные события в одну partition.
• Offset фиксирует прогресс consumer group; ручной commit после успешной обработки позволяет повторить незавершённую работу.

Частые ошибки

• Не используй автоcommit до завершения бизнес-обработки, если потеря сообщения недопустима.
• Увеличение числа partitions может менять распределение ключей и ограничивается числом параллельных consumers в группе.
• Retry и DLQ меняют порядок исходного потока и требуют явной политики возврата.

cfg := kafka.ReaderConfig{
    Brokers: []string{"kafka:9092"},
    GroupID: "orders-workers",
    Topic: "orders",
}
r := kafka.NewReader(cfg)
// Этот код запущен в каждом из 8 экземпляров.

m, err := r.ReadMessage(ctx)
if err != nil { return err }
return chargeDB(ctx, m)