Apache Kafka в Go

Зачем знать: Kafka — стандарт для event streaming, audit logs, CDC, аналитики, межсервисной асинхронной коммуникации. В 2026 middle 1 Go-разработчик обязан понимать topic/partition/offset, разницу между producer ack-уровнями, consumer group rebalancing, идемпотентного producer-а, схемы (Avro/Protobuf) и DLQ-паттерн. Без этого ваш сервис либо теряет сообщения, либо дублирует, либо превращает Kafka в очередь медленных задач.

1. Базовая концепция

Что такое Kafka

Apache Kafka — distributed append-only commit log. Producer пишет сообщения в topic, разбитый на partitions. Каждое сообщение получает offset (порядковый номер внутри partition). Consumer читает сообщения по offsets.

Главное отличие от очередей (RabbitMQ): сообщение не удаляется после чтения. Хранится по retention policy (по времени или размеру). Можно перечитать (replay).

Ключевые концепции

Broker — узел Kafka, хранит partitions и обслуживает запросы.
Cluster — набор брокеров с координацией через KRaft (в 2026, Zookeeper deprecated).
Topic — логическое имя потока. Например, orders.created.
Partition — физический shard topic. Сообщения внутри partition упорядочены, между partitions — нет.
Offset — уникальный номер сообщения внутри partition.
Leader/Replica — каждая partition имеет один leader и N replicas (на разных брокерах). ISR (In-Sync Replicas) — replicas, догнавшие leader.
Producer — пишет сообщения, выбирает partition по ключу/round-robin.
Consumer — читает сообщения, коммитит offset.
Consumer Group — набор consumers, делящих partitions topic-а. Один partition = один consumer в группе.
Rebalancing — при подключении/отключении consumer-а partitions перераспределяются.
Schema Registry — отдельный сервис для хранения схем (Avro/Protobuf/JSON).

Acks (durability vs latency)

acks=0 — fire-and-forget. Самое быстрое, можно потерять сообщение.
acks=1 — ждать ack от leader. Если leader умрёт до репликации, потеря.
acks=all (или -1) — ждать ISR. Самое надёжное.

С acks=all + min.insync.replicas=2 + replication.factor=3 → потеря только при одновременном падении всего кластера.

Идемпотентный producer

enable.idempotence=true — каждое сообщение получает sequence number, broker dedup-ит дубли при ретраях. Без этого ретрай может создать дубликат.

Exactly-once

Transactional producer + consumer с isolation.level=read_committed → exactly-once семантика (E2E). Реальное применение редко — большинство систем on at-least-once + idempotent consumer.

Compression

none — без сжатия.
gzip — лучшее сжатие, медленно (CPU).
snappy — быстро, среднее сжатие.
lz4 — баланс (дефолт в 2026).
zstd — лучшее сжатие/CPU ratio (рекомендуется в 2026).

Сжимаются batches, не отдельные сообщения.

Partitioning

Producer выбирает partition:

По key (hash(key) % num_partitions) — гарантирует порядок для key.
Round-robin без key.
Кастомный partitioner.

Retention

retention.ms — сколько хранить (default 7 дней).
retention.bytes — макс размер.
Log compaction — для key-value: оставляется последняя версия каждого key (для CDC, state stores).

Consumer group rebalancing

Когда consumer присоединяется/уходит, coordinator перераспределяет partitions. Алгоритмы:

Range — partitions подряд (старый default).
Round-robin.
Sticky — минимально меняет назначение (рекомендуется).
Cooperative Sticky — без stop-the-world rebalance (2.4+, рекомендуется).

Альтернативы

Redpanda — Kafka-compatible, написан на C++/Seastar, без JVM. Часто для dev/прод среднего масштаба.
Apache Pulsar — multi-tenancy, tiered storage.
AWS Kinesis — managed Kafka-like от AWS.

2. Как в Go (с примерами)

2.1 Клиенты

Клиент	Тип	Pro	Cons
`confluent-kafka-go`	CGO (librdkafka)	Best perf, polный набор фич	CGO, проблемы в Alpine
`segmentio/kafka-go`	pure Go	Простой, идиоматичный	Меньше фич, медленнее
`IBM/sarama` (Shopify)	pure Go	Старый, много примеров	Сложный API, рефакторинг продолжается
`twmb/franz-go`	pure Go	Modern, fast, exactly-once	Молодой (но в 2024-2026 — рекомендация)

В 2026 рекомендация: twmb/franz-go для большинства новых проектов, confluent-kafka-go если уже используется и CGO не проблема.

2.2 Установка franz-go

go get github.com/twmb/franz-go
go get github.com/twmb/franz-go/pkg/kgo

2.3 Producer (franz-go)

package main

import (
  "context"
  "log"

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

func main() {
  cl, err := kgo.NewClient(
    kgo.SeedBrokers("kafka:9092"),
    kgo.DefaultProduceTopic("orders.created"),
    kgo.RequiredAcks(kgo.AllISRAcks()),
    kgo.ProducerBatchCompression(kgo.ZstdCompression()),
    kgo.ProducerLinger(10*time.Millisecond),
    kgo.MaxBufferedRecords(10_000),
    // идемпотентность включена по умолчанию
  )
  if err != nil {
    log.Fatal(err)
  }
  defer cl.Close()

  ctx := context.Background()
  record := &kgo.Record{
    Key:   []byte("user-123"),
    Value: []byte(`{"order_id":"o-1","amount":100}`),
    Headers: []kgo.RecordHeader{
      {Key: "trace_id", Value: []byte("abc-def")},
    },
  }

  // Sync produce
  res := cl.ProduceSync(ctx, record)
  for _, r := range res {
    if r.Err != nil {
      log.Printf("produce error: %v", r.Err)
    }
  }

  // Async produce (callback)
  cl.Produce(ctx, record, func(r *kgo.Record, err error) {
    if err != nil {
      log.Printf("async produce error: %v", err)
    }
  })
}

2.4 Consumer (franz-go)

func consumer() {
  cl, err := kgo.NewClient(
    kgo.SeedBrokers("kafka:9092"),
    kgo.ConsumerGroup("orders-workers"),
    kgo.ConsumeTopics("orders.created"),
    kgo.DisableAutoCommit(),       // commit вручную
    kgo.SessionTimeout(30*time.Second),
    kgo.RebalanceTimeout(60*time.Second),
    kgo.FetchMaxBytes(50 * 1024 * 1024),
  )
  if err != nil {
    log.Fatal(err)
  }
  defer cl.Close()

  ctx := context.Background()
  for {
    fetches := cl.PollFetches(ctx)
    if errs := fetches.Errors(); len(errs) > 0 {
      for _, e := range errs {
        log.Printf("fetch error topic=%s part=%d: %v", e.Topic, e.Partition, e.Err)
      }
      continue
    }

    fetches.EachRecord(func(r *kgo.Record) {
      if err := processOrder(r.Key, r.Value); err != nil {
        log.Printf("process error: %v", err)
        // можно отправить в DLQ, см. ниже
        return
      }
    })

    if err := cl.CommitUncommittedOffsets(ctx); err != nil {
      log.Printf("commit error: %v", err)
    }
  }
}

func processOrder(k, v []byte) error {
  // ...
  return nil
}

2.5 Consumer с batch-обработкой

fetches.EachPartition(func(p kgo.FetchTopicPartition) {
  records := p.Records
  if len(records) == 0 {
    return
  }
  // обработка батча — выгоднее, можно делать insert N rows один раз
  if err := processBatch(records); err != nil {
    // при ошибке: пересоздать consumer или skip и в DLQ
    return
  }
})

2.6 Manual offset commit

cl, _ := kgo.NewClient(..., kgo.DisableAutoCommit())

for {
  fetches := cl.PollFetches(ctx)
  // обработка
  cl.CommitRecords(ctx, fetches.Records()...) // commit только успешно обработанных
}

CommitUncommittedOffsets коммитит последние offset-ы всех partitions. CommitRecords — конкретные.

2.7 confluent-kafka-go пример (для сравнения)

import "github.com/confluentinc/confluent-kafka-go/v2/kafka"

p, _ := kafka.NewProducer(&kafka.ConfigMap{
  "bootstrap.servers":   "kafka:9092",
  "acks":                "all",
  "enable.idempotence":  true,
  "compression.type":    "zstd",
})

deliveryChan := make(chan kafka.Event, 1000)
p.Produce(&kafka.Message{
  TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
  Key:            []byte("user-123"),
  Value:          []byte("payload"),
}, deliveryChan)

e := <-deliveryChan
m := e.(*kafka.Message)
if m.TopicPartition.Error != nil {
  // error
}

2.8 Schema Registry (Avro)

import "github.com/heetch/avro" // или confluent SR client

schema := `{"type":"record","name":"Order","fields":[
  {"name":"order_id","type":"string"},
  {"name":"amount","type":"long"}
]}`

codec, _ := avro.ParseSchema(schema)

binData, _ := codec.Marshal(Order{OrderID: "o-1", Amount: 100})

// + 5 байт: magic byte (0) + schema id (4 bytes)
header := []byte{0, 0, 0, 0, byte(schemaID)}
payload := append(header, binData...)

record := &kgo.Record{Value: payload}

В реале используют SR-клиент (Confluent SR, Karapace), который кеширует схемы и сериализует автоматически.

2.9 Protobuf

syntax = "proto3";
message Order {
  string order_id = 1;
  int64 amount = 2;
}

order := &Order{OrderID: "o-1", Amount: 100}
data, _ := proto.Marshal(order)

record := &kgo.Record{Value: data}

Protobuf чаще выбирается в Go-командах (тот же .proto для gRPC).

2.10 DLQ (Dead Letter Queue)

const dlqTopic = "orders.created.dlq"

func process(ctx context.Context, cl *kgo.Client, r *kgo.Record) {
  if err := doWork(r); err != nil {
    // в DLQ с info об ошибке
    dlq := &kgo.Record{
      Topic: dlqTopic,
      Key:   r.Key,
      Value: r.Value,
      Headers: append(r.Headers,
        kgo.RecordHeader{Key: "error", Value: []byte(err.Error())},
        kgo.RecordHeader{Key: "origin_topic", Value: []byte(r.Topic)},
        kgo.RecordHeader{Key: "origin_partition", Value: []byte(fmt.Sprint(r.Partition))},
        kgo.RecordHeader{Key: "origin_offset", Value: []byte(fmt.Sprint(r.Offset))},
      ),
    }
    cl.ProduceSync(ctx, dlq)
  }
}

DLQ — отдельная topic для сообщений, которые не удалось обработать. Можно потом перечитать и retry.

2.11 Retry с exponential backoff

const maxRetries = 5

for attempt := 0; attempt < maxRetries; attempt++ {
  if err := process(r); err == nil {
    break
  }
  time.Sleep(time.Duration(1<<attempt) * time.Second)
}

Или используйте delayed retry topic-и: orders.retry.5s, orders.retry.30s, orders.retry.5m (consumer ждёт).

2.12 Exactly-once (transactional)

cl, _ := kgo.NewClient(
  kgo.SeedBrokers("kafka:9092"),
  kgo.TransactionalID("payments-1"),
)
cl.BeginTransaction()

cl.Produce(ctx, record, nil)
cl.Produce(ctx, anotherRecord, nil)

if err := cl.EndAndBeginTransaction(ctx, kgo.TryCommit, kgo.EndBeginTxnSafe); err != nil {
  // rollback
}

Кейсы: read from topic A, transform, write to topic B атомарно (например, Kafka Streams pattern).

2.13 OpenTelemetry интеграция

import "github.com/twmb/franz-go/plugin/kotel"

tracer := kotel.NewTracer(kotel.TracerProvider(otel.GetTracerProvider()))
meter := kotel.NewMeter()
kotelOptions := kotel.NewKotel(kotel.WithTracer(tracer), kotel.WithMeter(meter))

cl, _ := kgo.NewClient(
  kgo.SeedBrokers("kafka:9092"),
  kgo.WithHooks(kotelOptions.Hooks()...),
)

Каждое produce/consume создаёт spans, headers содержат traceparent.

2.14 Local dev: redpanda

services:
  redpanda:
    image: redpandadata/redpanda:latest
    command:
      - redpanda
      - start
      - --smp 1
      - --memory 1G
      - --reserve-memory 0M
      - --kafka-addr 0.0.0.0:9092
      - --advertise-kafka-addr localhost:9092
    ports:
      - "9092:9092"

Redpanda Kafka-compatible, не нужна JVM, быстрый старт.

2.15 Health check

Простой ping: list brokers.

admin := kadm.NewClient(cl)
_, err := admin.ListBrokers(ctx)

3. Gotchas

3.1 acks=1 теряет сообщения

Если leader umer до репликации — данные потеряны. Для важных событий всегда acks=all + min.insync.replicas=2 + replication.factor=3.

3.2 Auto-commit + crash

EnableAutoCommit=true // default

Auto-commit фиксирует offset каждые 5 секунд. Если crash после обработки 4 секунды сообщений, но до commit — обработаются повторно (at-least-once). Если crash после commit, но до обработки — потеряются (at-most-once).

В prod: manual commit после успешной обработки.

3.3 Дубликаты при retry

Без enable.idempotence=true retry producer-а создаёт дубли. Включайте идемпотентность всегда.

3.4 Порядок сообщений

Порядок гарантирован только внутри partition. Если для одного user_id важен порядок, ключ = user_id → попадёт в один partition.

3.5 max.in.flight.requests.per.connection

Если > 1 и нет идемпотентности — порядок может нарушиться при retry. С идемпотентностью можно до 5 (default).

3.6 Rebalancing блокирует обработку

При rebalance (consumer joined/left) все consumers замораживаются (stop-the-world). С cooperative-sticky — только затронутые partitions.

3.7 Большие сообщения

Default max message size = 1MB. Большие → S3 + Kafka stores только URL. Не пытайтесь увеличить лимит до 10MB+ — это убьёт latency.

3.8 Consumer lag

consumer_lag = log_end_offset - consumer_offset — сколько сообщений ждёт. Если лаг растёт — consumer медленнее producer-а. Алертить.

3.9 Слишком много partitions

Каждая partition — файлы на диске, реплики, открытые fd. > 4000 partitions на брокер = деградация. Не делайте partition=1000 «на запас».

3.10 Слишком мало partitions

Один partition = один consumer в группе. Если хочется параллелить — нужно partitions >= consumers.

3.11 Schema evolution

Backward compatible (новые consumers с старыми messages): добавить optional поле OK. Forward compatible (старые consumers с новыми): не добавлять required. SR enforces.

3.12 enable.idempotence требует acks=all

Без acks=all идемпотентность невозможна — broker не подтвердит durability.

3.13 Consumer не делает heartbeat — kicked

Default session.timeout.ms=10s. Если обработка одного сообщения дольше → consumer считается dead, rebalance. Увеличьте max.poll.interval.ms (default 5 минут).

3.14 Poll loop блокирует

for {
  fetches := cl.PollFetches(ctx) // блокируется до сообщения
  // если обработка в горутине без back-pressure — OOM
}

Контролируйте конкуренцию. Worker pool с bounded channel.

3.15 Manual commit не атомарен с processing

process(r)
commit(r) // crash между этими двумя → at-least-once

Идемпотентный consumer обязателен. Или transactional producer (для read-process-write).

3.16 ConsumerGroup с одинаковым ID

Два инстанса с одинаковым group.id — они делят partitions. Если хотите два независимых стрима — разные group.id.

3.17 Topic auto-create

В prod лучше отключать auto.create.topics.enable=false. Иначе опечатка в имени = новый topic с дефолтными настройками (1 replica, retention 7d).

3.18 Compression на producer != decompress на consumer overhead

Compression уменьшает сеть, но требует CPU на decompress. Для CPU-bound сервиса баланс важен.

4. Best practices в production

4.1 Producer

kgo.RequiredAcks(kgo.AllISRAcks())
// идемпотентность включена по умолчанию в franz-go
kgo.ProducerBatchCompression(kgo.ZstdCompression())
kgo.ProducerLinger(10 * time.Millisecond)         // батчинг
kgo.ProducerBatchMaxBytes(1024 * 1024)             // 1MB max batch
kgo.MaxBufferedRecords(10_000)
kgo.ProduceRequestTimeout(30 * time.Second)
kgo.RecordDeliveryTimeout(60 * time.Second)

4.2 Consumer

kgo.ConsumerGroup("my-group")
kgo.ConsumeTopics("orders.created")
kgo.DisableAutoCommit()
kgo.SessionTimeout(30 * time.Second)
kgo.RebalanceTimeout(60 * time.Second)
kgo.HeartbeatInterval(3 * time.Second)
kgo.FetchMaxBytes(50 * 1024 * 1024)
kgo.FetchMaxWait(500 * time.Millisecond)

4.3 Topic config

replication.factor=3 (HA на потерю одной ноды).
min.insync.replicas=2 (требует, чтобы 2 из 3 реплик подтвердили).
cleanup.policy=delete (или compact для CDC).
retention.ms=604800000 (7 дней).
segment.bytes=1073741824 (1GB segment files).
compression.type=producer (broker не пережимает).

4.4 Partitions: правильное число

partitions = max(target_throughput / partition_throughput, target_consumers)

Обычно 12-100 на topic. Можно увеличить позже, но hash(key) % partitions поменяется → порядок ломается на старых данных.

4.5 Key для порядка

Если важен порядок по user_id, всегда key = user_id. Если порядок не важен — key пустой (round-robin распределение).

4.6 Schema Registry — обязательно

Без схемы producer пишет, что хочет, consumer падает. SR enforces compatibility.

4.7 DLQ + retry topics

Pattern:

orders.created (main)
orders.created.retry (consumer ждёт N секунд)
orders.created.dlq (manual triage)

При ошибке → retry. После K попыток → DLQ.

4.8 Idempotent consumer

Обработка сообщения должна быть идемпотентной (дубли не ломают state):

INSERT ON CONFLICT DO NOTHING.
Outbox: уникальный event_id, отвергаем повторы.
Storage с natural idempotency (timeseries upsert).

4.9 Outbox pattern

Запись в БД + event в Kafka должны быть атомарны. Решение:

В одной транзакции с бизнес-данными INSERT в outbox таблицу.
Отдельный процесс (Debezium CDC или scheduled job) читает outbox и пишет в Kafka.

4.10 Tracing

См. 2.13. Trace headers (traceparent) автоматически. Каждое produce/consume — span.

4.11 Metrics

kafka_consumer_lag — алерт > 10K.
kafka_consumer_records_consumed_rate.
kafka_producer_record_error_rate.

4.12 Graceful shutdown

signal.NotifyContext(...)
cl.Close() // дожидается inflight produces, leaves consumer group

Без этого rebalance не triggers сразу — другие consumers ждут session.timeout.

4.13 Не использовать Kafka как RPC

Kafka — async event log. Если нужен sync request/response — gRPC/HTTP, не Kafka.

4.14 Не использовать Kafka как очередь задач

Kafka не имеет per-message ack/priority/delay. Для job queue — RabbitMQ, NATS JetStream или Redis Streams.

4.15 Monitoring lag и offsets

kafka-consumer-groups.sh --bootstrap-server kafka:9092 --describe --group my-group

Или Burrow, Cruise Control, kafka_exporter для Prometheus.

4.16 Storage backups

Mirror Maker 2 / Cluster Linking / S3 sink для disaster recovery.

4.17 Authentication

SASL (SCRAM-SHA-256/512) + TLS. Не plain text. ACLs на topic.

4.18 Не используйте `auto.offset.reset=latest` для критичных

Default latest означает «при первом запуске начни с конца» — пропустишь старые. Используйте earliest для аналитики/полной обработки.

4.19 Quotas

Producer/consumer quotas защищают от noisy neighbour: limit MB/s по user/clientid.

4.20 Партиции должны быть осмысленные

Если 99% трафика идёт под одним ключом — все попадают в одну partition (hot partition). Решение: добавьте suffix к ключу userID + bucket(time) или hash(userID) % N.

5. Вопросы на собесе

Что такое Kafka в одном предложении? Distributed, persistent, partitioned commit log с pub-sub семантикой.
Topic vs partition? Topic — логическое имя, partition — физический shard. Сообщения упорядочены внутри partition.
Что такое offset? Уникальный последовательный номер сообщения в partition. Consumer хранит свой текущий offset.
acks=0/1/all? acks=0 — fire-forget. acks=1 — ждать leader. acks=all — ждать ISR. Для durability — all.
Что такое ISR? In-Sync Replicas — реплики, которые догнали leader (lag < threshold). При acks=all нужны N ISR (min.insync.replicas).
Idempotent producer — зачем? Без него retry создаёт дубликаты (broker не знает, что producer уже отправлял). С ним broker dedup-ит по sequence number.
Что такое consumer group? Группа consumers, делящих partitions topic. Один partition = один consumer в группе. Масштабирование через увеличение consumers до количества partitions.
Что происходит при rebalance? Coordinator перераспределяет partitions между consumers. Stop-the-world (старый), или cooperative-sticky (без полной остановки, 2.4+).
At-most-once vs at-least-once vs exactly-once?
- At-most-once: commit перед обработкой → может потеряться.
- At-least-once: обработка → commit → может дублироваться.
- Exactly-once: transactional producer + consumer (read_committed).
Как обеспечить порядок сообщений? Ключ = идентификатор сущности (user_id, order_id). Все сообщения с одним ключом → одна partition → упорядочены.
Что такое log compaction? Cleanup-policy compact оставляет только последнее значение для каждого key. Для state stores, CDC, snapshots.
DLQ pattern? Сообщения, которые consumer не смог обработать после retry → отдельный topic для ручного разбора.
Чем Kafka отличается от RabbitMQ? Kafka — persistent log, replay, high throughput, ordering, NO per-message ack/priority. RabbitMQ — flexible routing, per-message ack, smaller scale.
Какие Go-клиенты? confluent-kafka-go (CGO), segmentio/kafka-go, IBM/sarama, twmb/franz-go (новый рекомендуемый в 2026).
Что такое Schema Registry? Сервис для хранения схем (Avro, Protobuf, JSON Schema). Producer и consumer регистрируют/получают схемы. SR enforces compatibility.
Backward vs forward compatibility? Backward — новый consumer может прочитать старые сообщения. Forward — старый consumer может прочитать новые. Full — оба.
Что такое consumer lag? log_end_offset - consumer_offset. Сколько сообщений ждёт обработки. Алертить если растёт.
Как увеличить throughput producer? Batching (linger.ms), compression, увеличить буфер, multiple producers.
Что такое replication.factor? Сколько копий каждой partition. 3 — стандарт (потеря одной ноды без потери данных).
Чем KRaft отличается от Zookeeper? KRaft — встроенный consensus в Kafka, заменил Zookeeper с 3.3+. Production-ready с 3.5 (2023). В 2026 — стандарт.
Что такое sticky vs cooperative-sticky rebalancing? Sticky минимизирует перемещения partitions. Cooperative-sticky дополнительно не останавливает все consumers (рекомендуется в 2026).
Когда не использовать Kafka? Sync RPC (используйте gRPC), low latency real-time (NATS), small scale (RabbitMQ), per-message delay/priority (RabbitMQ/SQS).
Что такое outbox pattern? Атомарность БД-транзакции и publish в Kafka. INSERT в outbox в одной транзакции с бизнес-данными, отдельный процесс читает outbox → Kafka.
Чем Redpanda отличается от Kafka? Wire-compatible, без JVM (C++/Seastar), faster startup, lower latency. Для multi-region / cloud-native часто выбирается.
Как отлаживать lag? kafka-consumer-groups --describe, Burrow, kafka_exporter + Grafana. Смотреть lag по partition, выявить медленный consumer.
Какой compression лучше в 2026? zstd — лучший ratio/CPU. lz4 — быстрее, меньше CPU. Snappy — старый default.
Что такое min.insync.replicas? Минимум ISR для acks=all. min.insync.replicas=2 с replication.factor=3 → отказ одной ноды OK, две — ошибка produce.
Зачем нужны session.timeout.ms и heartbeat.interval.ms? Heartbeat периодически шлёт coordinator-у. Если нет heartbeat за session.timeout — consumer считается мертвым, rebalance.
Партиция = 1 consumer. Что если consumers > partitions? Лишние consumers idle. Увеличьте partitions (но не уменьшите потом).
Что такое exactly-once в Kafka Streams? processing.guarantee=exactly_once_v2. Transactional producer + consumer + state store с idempotent updates.

6. Practice

Упражнение 1: Простой producer/consumer

С помощью franz-go: producer пишет 1000 сообщений в topic events, consumer читает и печатает. Запустите Redpanda локально.

Упражнение 2: Producer config

Включите идемпотентность, acks=all, zstd compression, batch linger 50ms. Проверьте throughput через pprof.

Упражнение 3: Consumer group

Запустите 3 инстанса consumer-а с одним group.id. Создайте topic с 6 partitions. Убедитесь, что каждый получит по 2 partition. Убейте один — rebalance.

Упражнение 4: Manual commit

Реализуйте consumer с manual commit после успешной обработки. При ошибке — не коммитить, retry на следующем poll.

Упражнение 5: DLQ

При ошибке обработки сообщение → DLQ topic с headers (origin, error). Напишите отдельный consumer для DLQ.

Упражнение 6: Schema Registry

Используйте Avro схему для OrderCreated. Регистрируйте через SR (Confluent или Karapace в docker-compose). Producer сериализует, consumer десериализует автоматически.

Упражнение 7: Tracing

Подключите kotel (или ручной propagation через headers). Запустите producer + consumer, в Jaeger увидьте trace с двумя spans.

Упражнение 8: Outbox pattern

В PostgreSQL транзакции: INSERT в orders + INSERT в outbox_events. Отдельная горутина: SELECT FOR UPDATE SKIP LOCKED из outbox → produce → DELETE.

Упражнение 9: Хот partition

Симулируйте hot partition (90% сообщений с одним ключом). Покажите консьюмер-лаг на одной partition. Введите bucketing.

Упражнение 10: Exactly-once

Реализуйте read-process-write с transactional producer. Прочитайте из A, transform, запишите в B атомарно. Проверьте, что consumer B с read_committed не видит uncommitted сообщений.

7. Источники

Apache Kafka docs — https://kafka.apache.org/documentation/.
“Kafka: The Definitive Guide” by Gwen Shapira et al. (O’Reilly, 2nd edition, 2021).
franz-go — https://github.com/twmb/franz-go.
confluent-kafka-go — https://github.com/confluentinc/confluent-kafka-go.
Confluent blog — https://www.confluent.io/blog/.
Redpanda docs — https://docs.redpanda.com/.
“Designing Data-Intensive Applications” by Martin Kleppmann (O’Reilly, 2017) — главы 11 (streaming).
Kafka improvement proposals (KIPs) — https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Improvement+Proposals (для свежих фич).
Schema Registry docs (Confluent) — https://docs.confluent.io/platform/current/schema-registry/.
Debezium (CDC) — https://debezium.io/ (Outbox connector).