Модуль 1. Apache Kafka — поток, который не остановить

Цель модуля: научиться думать «логом»: topic/partition/offset вместо очередей, консьюмер-группы вместо балансировщиков, повторное чтение вместо «сообщение сгорело». Освоить продюсера с батчингом и acks, перебалансировки, транзакции, compacted topics, Schema Registry, интеграцию с базой и KRaft — и собрать CQRS-лабораторную «Event-sourced счёт».

Перед началом: курс «Фундамент» (Модуль 1). Понимание at-least-once и идемпотентности обязательно. Инструменты: docker-compose (apache/kafka:3.7.0, KRaft), segmentio/kafka-go, franz-go (обзорно), PostgreSQL. Трудоёмкость: ~14–17 ч (теория 6 ч · практика 8 ч · квиз 1 ч).

📖 1.1. Модель: topic, partition, offset, consumer group

Лог, а не очередь

Kafka — это append-only журнал (лог), разбитый на партиции:

text
topic "orders-events"
partition 0: [0][1][2][3][4][5] ...
partition 1: [0][1][2][3][4] ...
partition 2: [0][1][2][3][4][5][6] ...

• Запись — всегда в конец партиции. Сообщение внутри партиции имеет монотонный offset.
• Порядок существует только внутри партиции. Между партициями — нет.
• Сообщения не удаляются после чтения: они живут по retention (времени/размеру), и каждый консьюмер читает свой прогресс.

Consumer group — группа инстансов одного сервиса. Партиции распределяются между членами группы: каждая партиция в каждый момент читается ровно одним членом группы. Смещение прочитанного (committed offset) группа хранит внутреннем топике __consumer_offsets.

text
consumer group "payment"
partition 0 ───────────────────────────► [consumer-1]
partition 1 ───────────────────────────► [consumer-2]
partition 2 ───────────────────────────► [consumer-3]

Отсюда два следствия, которые нужно назвать до того, как их спросят:

1. Параллелизм группы = число партиций. 5 инстансов на 3 партиции — два инстанса простаивают (и это не баг).
2. Глобальный порядок в топике с 2+ партициями невозможен. Хотите порядок по сущности — кладите её события в одну партицию одним и тем же ключом (см. 3.2).

Выбор библиотеки: честная таблица

В уроках используем segmentio/kafka-go — на нём видно все примитивы. Но когда дойдём до exactly-once (3.4), честно скажем: транзакции в нём не поддерживаются, и для продакшен-EOS берут franz-go или librdkafka. Это типичная цена «простой чисто-Go библиотеки».

Читаем и пишем на kafka-go

go
package main

import (
	"context"
	"log"
	"time"

	"github.com/segmentio/kafka-go"
)

func main() {
	w := &kafka.Writer{
		Addr:         kafka.TCP("localhost:9092"),
		Topic:        "orders-events",
		Balancer:     &kafka.Hash{}, // партиция по ключу — гарантия порядка по key
		RequiredAcks: kafka.RequireAll,
		BatchTimeout: 10 * time.Millisecond,
	}

	err := w.WriteMessages(context.Background(), kafka.Message{
		Key:   []byte("order-42"), // все события одного заказа — в одну партицию
		Value: []byte(`{"type":"created","amount":100}`),
	})
	if err != nil {
		log.Fatal(err)
	}

	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers: []string{"localhost:9092"},
		GroupID: "payment",        // включаем consumer group
		Topic:   "orders-events",
		MinBytes: 1,               // приходить сразу
		MaxBytes: 10e6,            // 10MB — максимальный размер фетча
	})
	for {
		m, err := r.FetchMessage(context.Background())
		if err != nil {
			log.Fatal(err)
		}
		log.Printf("partition=%d offset=%d key=%s", m.Partition, m.Offset, m.Key)
		r.CommitMessages(context.Background(), m) // сдвигаем committed offset
	}
}

📖 1.2. Продюсер: партицирование, батчинг, сжатие, acks

Ключ решает порядок

Балансер kafka.Hash берёт хэш ключа и выбирает партицию детерминированно: все события order-42 попадают в одну партицию → читаются в порядке записи. Ключ — это бизнес-решение, а не техническая деталь: сущность, по которой нужен порядок (order_id, account_id, user_id).

⚠️ Ловушка: при добавлении партиций маппинг key→partition меняется. Старые события ключа лежат в партиции 2, новые уйдут в партицию 5 — глобальный порядок по ключу сломан. Увеличивайте партиции либо до запуска, либо осознавая разрыв (или используйте ключи с встроенным временем).

Батчинг: главный рычаг throughput

Продюсер копит сообщения в батч на партицию и отправляет пачкой. Ручки:

Батчинг + сжатие — причина, по которой Kafka на больших потоках обходится меньшим числом запросов, чем per-message-брокеры: сеть тратится на байты, а не на фреймворк-оверхед.

acks: три позиции надёжности

acks=all работает в паре с min.insync.replicas (broker-настройка): при acks=all, min.insync.replicas=2 запись требует лидера + минимум одной реплики. Если ISR сжался до 1 — продюсер получает NOT_ENOUGH_REPLICAS и (правильно!) не пишет. Это не баг, а защита от записи «в одиночку», которую потом некому подтвердить.

При acks=all ретраи безопасны только с идемпотентным продюсером — об этом в 3.4. Без него retry после таймаута может задублировать сообщение (брокер записал, но ответ не дошёл).

📖 1.3. Консьюмер: rebalance, commit strategies

Перебалансировка: когда расклад партиций меняется

Группа перебалансируется, когда: член присоединился/умер (heartbeat-таймаут, session.timeout), консьюмер перестал делать poll (max.poll.interval), изменилось число партиций.

• Eager (по умолчанию в старых стратегиях): все консьюмеры сначала отдают все партиции (stop-the-world), потом получают новые назначения. Короткая полная пауза во всей группе.
• Cooperative-sticky: перераспределяются только «переезжающие» партиции, остальные продолжают читать. В kafka-go нет — только range/roundrobin. Для cooperative берите franz-go/librdkafka (назначается CooperativeStickyBalancedAssignmentStrategy).

Что обязан делать ваш код: обрабатывать revoke-события — остановить обработку партиций, которые забрали, закоммитить прогресс. Иначе партиция читается двумя консьюмерами одновременно.

Commit strategies: где сдвигать оффсет

В kafka-go: CommitInterval: 0 в конфиге ридера — ручной режим, r.CommitMessages(ctx, m) после успешной обработки. Коммитить можно пачкой: обработали батч из FetchMessage-ов — один commit последнего оффсета.

«Потерянные» оффсеты в реальной жизни: консьюмер упал после обработки, но до коммита → сообщение придёт снова. Это норма at-least-once, и весь ваш обработчик обязан это переживать идемпотентностью (Модуль 1 курса «Фундамент», раздел 1.7).

📖 1.4. Exactly-once: идемпотентный продюсер, транзакции, ограничения

Уровень 1: идемпотентный продюсер

Брокер дедуплицирует записи продюсера: у каждого продюсера есть PID, у каждого сообщения — последовательный номер по партиции. При ретрае брокер видит повтор sequence и отбрасывает дубль.

• Гарантия: без дублей при ретраях, в пределах сессии продюсера (PID меняется при рестарте — история обнуляется) и в разрезе партиции.
• Не решает: дубли между сессиями, дубли вашего приложения (дважды вызвали WriteMessages), порядок между партициями.

В kafka-go этого уровня нет — ещё один довод в пользу franz-go для серьёзного продакшена.

Уровень 2: транзакции

Транзакционный продюсер атомарно пишет в несколько партиций и помечает группу сообщений как committed/aborted. Консьюмеры с isolation.level=read_committed не видят aborted-сообщения. Классический use-case — consume-process-produce:

text
read(msg из topic-in) → обработка → write(в topic-out) + commit(offset) — ОДНОЙ транзакцией

Zombie fencing: при перезапуске с тем же transactional.id старый «зомби» продюсер фенсится (эпоха увеличивается), его записи отвергаются.

Честные ограничения

1. Транзакции Kafka не покрывают внешние системы. Записать в Kafka и в PostgreSQL атомарно транзакцией нельзя — для этого существует Outbox (Модуль 2 курса «Фундамент») и идемпотентный sink (1.7).
2. Латентность транзакций значима: commit + маркеры добавляют десятки мс. Хайлоад-стриминг часто живёт на at-least-once + идемпотентности.
3. Финальная правда: «exactly-once» в Kafka = идемпотентный продюсер + транзакции + read_committed внутри Kafka + идемпотентный консьюмер снаружи. Полный контур EOS соберём в курсе OrderFlow.

📖 1.5. Compacted topics, retention, log segments

Retention: лог как буфер времени

Партиция — файлы-сегменты (по умолчанию ~1 ГБ). Ротация сегмента — по размеру/времени; удаление — только целых закрытых сегментов, активный хвост не удаляется никогда. Отсюда практики: retention.ms меньше хвоста, который вы готовы перечитывать; следите за тем, что «сообщение удалено» означает «целый сегмент устарел», а не «сообщение пережило deadline».

Compacted topics: лог как таблица

cleanup.policy=compact — брокер хранит последнее значение по ключу:

text
key=u1, v=100   ← может быть вычищен
key=u2, v=200   ← может быть вычищен
key=u1, v=150   ← остаётся: последнее по ключу

• Полный лог с историей → компактед-лог «снимок на каждый момент старта».
• Так устроены __consumer_offsets и __cluster_metadata.
• Удаление ключа — tombstone (value=nil); после delete.retention.ms маркер вычищается. Не пишите tombstone «на всякий случай» — потребители обязаны уметь их трактовать.
• Компакт — не мгновенный: между записью и компакцией ключ встречается несколько раз, консьюмер обязан сам брать последнее (или читать с конца).

Когда Kafka — это база данных: compacted topic + KV-семантика хорош для «текущего состояния» в стриминге (state store, конфиги, changelog). Но: нет транзакций между ключами, нет вторичных индексов, нет SQL. Компромисс известен: Kafka хранит события, состояние живёт в базе/store, который стримится из лога (3.9).

📖 1.6. Schema Registry: Avro/Protobuf и эволюция схем

Зачем реестр схем

JSON-в-байтах без контракта через год превращается в «расшифруй это». Schema Registry хранит версии схем и отказывает в несовместимой эволюции: продюсер, регистрирующий ломающую схему, получает ошибку до того, как сломает консьюмеров.

Формат сообщения: [magic byte 0x00][4 байта schema ID][payload].

go
import "github.com/riferrei/srclient"

client := srclient.CreateSchemaRegistryClient("http://localhost:8081")
schema, err := client.GetLatestSchema("orders-events-value")
if err != nil {
	log.Fatal(err)
}
// schema.ID пишем в байты 1-4, payload кодируем по схеме (goavro для Avro,
// proto.Marshal для Protobuf)

Protobuf в Go-мире чаще Avro: кодогенерация из .proto, строгие типы, protojson. Avro исторически ближе к экосистеме Kafka. Оба работают с Registry; выбор чаще диктует организация.

Совместимость: BACKWARD / FORWARD / FULL

Почему «добавить поле с default» — золотое правило: старые консьюмеры прочитают новое сообщение (поля нет → default), новые — старое (поле применится). Ломают совместимость: переименование, смена типа, удаление без default у обязательного поля, смена семантики при том же имени.

Миграция: переименование user_id → customer_id за 3 деплоя

1. Деплой 1 (продюсер): пишем оба поля — user_id и customer_id с одинаковым значением. BACKWARD-совместимо.
2. Деплой 2 (все консьюмеры): читаем customer_id, на user_id не полагаемся. Прогоняем окна ретеншена.
3. Деплой 3 (продюсер): убираем user_id. Старых читателей больше нет.

Такой же протокол — для любых «необратимых» изменений схем. Разверните его в свой чек-лист: он понадобится в курсе OrderFlow на миграциях MVP.

📖 1.7. Kafka ↔ PostgreSQL: sink-паттерн и идемпотентная запись

Задача из реальной жизни: события из топика должны попасть в базу. Сбой в любом месте даёт повторную доставку — значит, запись обязана быть идемпотентной. Два механизма на выбор:

Вариант A — таблица применённых событий (универсальный):

sql
-- одна транзакция на событие
BEGIN;
INSERT INTO applied_events (event_id) VALUES ($1);          -- уникальный PK
  ON CONFLICT DO NOTHING;
-- если вставилось 0 строк — событие уже применено, COMMIT и идём дальше

UPDATE balances SET balance = balance + $2 WHERE account_id = $3;
COMMIT;

Вариант B — монотонная версия в строке (компактнее):

sql
INSERT INTO balances (account_id, balance, version)
VALUES ($1, $2, $3)
ON CONFLICT (account_id) DO UPDATE
SET balance = balances.balance + EXCLUDED.balance,
    version  = EXCLUDED.version
WHERE balances.version < EXCLUDED.version;  -- дубль старой версии не применится

Требование для B: у событий по каждому ключу монотонная версия (offset подходит, если проекция читает партицию последовательно).

Go-код sink-консьюмера (скелет):

go
for {
	m, err := r.FetchMessage(ctx)
	if err != nil {
		return err
	}
	ev, err := decode(m.Value) // schema ID из заголовка, см. 3.6
	if err != nil {
		return fmt.Errorf("poison message: %w", err) // стоп группы — см. Модуль 2 курса «Фундамент»
	}
	if err := applyTx(ctx, db, ev); err != nil { // идемпотентная запись
		return err
	}
	if err := r.CommitMessages(ctx, m); err != nil {
		return err
	}
}

Обратите внимание на порядок: БД — источник истины применённости, коммит оффсета — лишь оптимизация. Если упадём между applyTx и CommitMessages, сообщение придёт снова, applied_events отбросит дубль. Обратный порядок (сначала commit) — потеря события.

Партиционирование топика по account_id даёт последовательное применение событий счёта — тот же принцип, что в лабораторной 3.9.

📖 1.8. KRaft: Kafka без ZooKeeper

Историческая архитектура: ZooKeeper хранил метаданные (топики, конфиги, ISR, контроллер), брокеры и ZK — два распределённых системы в одном кластере. Проблемы: двойная эксплуатация, лимит масштаба (~200k партиций), медленный failover контроллера (выборы через ZK).

KRaft (Kafka Raft) заменяет ZooKeeper: метаданные — тоже лог (__cluster_metadata), реплицируемый Raft-кворумом контроллеров. С 3.3 — production-ready, в 4.0 ZooKeeper-режим удалён полностью.

Что меняется для вас:

• Роли процессов: broker, controller или combined (для дев/среднего прода). В docker-compose трека: KAFKA_CFG_PROCESS_ROLES=broker,controller и KAFKA_CFG_CONTROLLER_QUORUM_VOTERS — это и есть KRaft.
• Масштаб: миллионы партиций, failover контроллера за миллисекунды (лог метаданных уже реплицирован).
• Эксплуатация: один протокол безопасности/инструменты; состояние кворума смотрится так:

bash
kafka-metadata-quorum --bootstrap-server localhost:9092 describe --status

Миграция ZK→KRaft (если вам достался старый кластер): поэтапно — добавить контроллеры KRaft в режиме миграции, двойная запись метаданных, переключение брокеров, демонтаж ZK. В новых проектах вопрос закрыт: разворачивайте сразу KRaft. Заметьте: RabbitMQ-кластер (2.5) и Kafka-KRaft решают одну задачу — консенсус метаданных, но Raft-логом, а не кворумом отдельной системы.

📖 1.9. Лабораторная: «Event-sourced счёт»

Собираем CQRS-пайплайн: команда меняет состояние через событие, проекция считает баланс, порядок гарантируется ключом.

text
[Command API] --outbox--> topic accounts.events (key=account_id)
                                   |
                                   v
                        [Projector (group=projection)]
                                   |
                                   v
                        PostgreSQL: balances (проекция)

Продюсер: команда → событие (через outbox из Модуля 2 курса «Фундамент», здесь упрощённо)

go
func (s *Service) Deposit(ctx context.Context, accountID string, cents int64) error {
	ev := Event{ID: uuid.NewString(), Type: "deposited", Account: accountID, Amount: cents}
	body, _ := json.Marshal(ev)
	return s.writer.WriteMessages(ctx, kafka.Message{
		Key:   []byte(accountID), // все события счёта — одна партиция = порядок
		Value: body,
	})
}

Проектор: применяем события строго по порядку партиции

go
r := kafka.NewReader(kafka.ReaderConfig{
	Brokers: []string{"localhost:9092"},
	GroupID: "projection",
	Topic:   "accounts.events",
})
for {
	m, err := r.FetchMessage(ctx)
	if err != nil {
		return err
	}
	var ev Event
	if err := json.Unmarshal(m.Value, &ev); err != nil {
		return fmt.Errorf("poison: %w", err)
	}
	if err := apply(ctx, s.db, ev); err != nil { // см. 3.7, вариант B
		return err
	}
	if err := r.CommitMessages(ctx, m); err != nil {
		return err
	}
}

sql
INSERT INTO balances (account_id, balance, version)
VALUES ($1, $2, $3)
ON CONFLICT (account_id) DO UPDATE
SET balance = balances.balance + EXCLUDED.balance, version = EXCLUDED.version
WHERE balances.version < EXCLUDED.version;

Что должно получиться:

1. Тест «100 конкурентных депозитов по 10 счётам» — итоговая сумма сходится до копейки, version каждого счёта = числу его событий.
2. Тест «краш проектора посреди батча» — повторная доставка не меняет финальную сумму (проверка идемпотентности).
3. Тест «два проектора в группе» — партиция одна, события счёта не задваиваются в проекции.

💻 Практика

1. lesson3_1: докажите тестом, что при 5 инстансах группы и 3 партициях два читателя получают ноль сообщений.
2. lesson3_2: снимите throughput продюсера при BatchTimeout 0/10/100 мс и сжатии none/gzip — объясните цифры.
3. lesson3_3: поднимите группу из 3 консьюмеров, убейте один — замерьте паузу потребления (eager) и повторные доставки.
4. lesson3_6: зарегистрируйте схему v1, попробуйте v2 с переименованием поля — Registry должен отказать; пройдите миграцию по трём деплоям из 3.6.
5. Промежуточный проект: лабораторная 3.9 целиком + графики consumer lag.

🧩 Quiz "Проверь себя"

✅ Итоги модуля

После прохождения ты должен уметь:

• Объяснить модель лога: почему порядок только в партиции и как ключ решает, где живёт порядок.
• Настроить продюсера: батчинг, сжатие, acks — и назвать цену каждой позиции acks вместе с min.insync.replicas.
• Управлять перебалансировками: eager vs cooperative, revoke-события, стратегии коммита и их семантики.
• Честно ответить, что такое exactly-once в Kafka (идемпотентный продюсер + транзакции + read_committed) и чего он не покрывает (внешние системы).
• Использовать compacted topics и tombstone'ы, объяснить retention в терминах сегментов.
• Эволюционировать схемы под контролем Registry (BACKWARD/FORWARD/FULL) и провести трёхдеплойную миграцию.
• Писать идемпотентный sink в PostgreSQL и знать, почему коммит оффсета идёт после записи в БД.
• Развернуть KRaft-кластер и объяснить, чем он лучше связки с ZooKeeper.

Шпаргалка модуля

🔗 Куда дальше

Kafka силён историей и порядком. Мы разобрали её модель от партиций до KRaft; продвинутые темы — мониторинг (lag, ISR, request latency), инциденты (rebalance storm, disk full) и тюнинг под нагрузкой — выходят следующими модулями этого курса. А поставить всё вместе можно уже сейчас: дальше в треке — курсы RabbitMQ и Redis как брокер, затем капстоун OrderFlow, который проведёт пайплайн через все брокеры сразу.