Kafka в production на Go: kafka-go, DLQ, Schema Registry, тюнинг, безопасность

← назад к разделу

Если вы уже знаете, что такое топик, партиция и группа потребителей — эта статья про следующий шаг: как подключить Kafka к Go-приложению и что нужно настроить, чтобы оно работало надёжно в production.

kafka-go: Writer и Reader

Библиотека segmentio/kafka-go — самый распространённый Kafka-клиент на чистом Go, без cgo. Для отправки сообщений используют Writer, для приёма — Reader и consumer loop.

// Отправка
type OrderEventPublisher struct {
	writer *kafka.Writer
}

func (p *OrderEventPublisher) Publish(ctx context.Context, event OrderEvent) error {
	payload, err := json.Marshal(event)
	if err != nil {
		return err
	}
	return p.writer.WriteMessages(ctx, kafka.Message{
		Key:   []byte(event.OrderID),
		Value: payload,
	})
}

// Приём
reader := kafka.NewReader(kafka.ReaderConfig{
	Brokers: []string{"localhost:9092"},
	Topic:   "orders",
	GroupID: "billing-service",
})

for {
	msg, err := reader.ReadMessage(ctx)
	if err != nil {
		break
	}
	var event OrderEvent
	if err := json.Unmarshal(msg.Value, &event); err != nil {
		continue
	}
	// обработка
}

Consumer loop крутится в одной goroutine. Чтобы обрабатывать партиции параллельно, запускают несколько goroutine — каждая со своим Reader и одним GroupID: Kafka распределит партиции между ними.

Commit offset: когда Kafka «знает», что сообщение обработано

Kafka следит за тем, до какого сообщения дошёл потребитель, через offset commit. В kafka-go момент коммита определяется способом чтения:

• ReadMessage — коммит сразу при чтении, до обработки. Просто, но при сбое между чтением и обработкой сообщение теряется.
• CommitInterval > 0 — коммит пачками в фоне раз в интервал. Быстрее, но при сбое часть сообщений придёт повторно.
• FetchMessage + CommitMessages — ручной коммит. Нужен, когда важно закоммитить оффсет только после успешной записи в базу.

reader := kafka.NewReader(kafka.ReaderConfig{
	Brokers:        []string{"localhost:9092"},
	Topic:          "orders",
	GroupID:        "billing-service",
	CommitInterval: 0, // 0 — синхронный коммит; >0 — пачками в фоне
})

for {
	msg, err := reader.FetchMessage(ctx)
	if err != nil {
		break
	}
	if err := handle(msg); err != nil {
		continue
	}
	reader.CommitMessages(ctx, msg) // только после успешной обработки
}

Заголовки сообщений

Каждое сообщение Kafka может содержать заголовки — пары (ключ, значение) в байтах, отдельно от полезной нагрузки. Сюда кладут техническую метаинформацию, которой не место в бизнес-данных:

• X-Correlation-ID / traceparent — для распределённого трейсинга.
• X-Event-Version — версия формата события.
• X-Source-Service — откуда пришло сообщение.

// При отправке
writer.WriteMessages(ctx, kafka.Message{
	Key:   []byte(key),
	Value: payload,
	Headers: []kafka.Header{
		{Key: "X-Correlation-ID", Value: []byte(correlationID)},
	},
})

// При получении
var correlationID string
for _, h := range msg.Headers {
	if h.Key == "X-Correlation-ID" {
		correlationID = string(h.Value)
	}
}
logger := slog.With("correlationId", correlationID)
// обработка

Бизнес-данные (orderId, amount, status) — в payload. Всё техническое — в headers.

Dead Letter Queue: что делать с проблемными сообщениями

Если обработчик возвращает ошибку и оффсет не коммитится, потребитель снова и снова возвращается к тому же сообщению — оффсет не двигается, потребитель встаёт. На production это означает полную остановку группы.

Решение — Dead Letter Queue (DLQ): после нескольких неудачных попыток сообщение перекладывается в отдельный топик (обычно с суффиксом .DLT), оффсет коммитится, основной потребитель продолжает работу.

Простой способ: retry на месте + DLQ

Готовой обвязки в kafka-go нет — DLQ собирается вручную из Reader, Writer и цикла повторов:

func consumeWithDLQ(ctx context.Context, r *kafka.Reader, dlq *kafka.Writer) {
	for {
		msg, err := r.FetchMessage(ctx)
		if err != nil {
			return
		}
		if err := processWithRetry(ctx, msg, 3, time.Second); err != nil {
			dlq.WriteMessages(ctx, kafka.Message{
				Key:   msg.Key,
				Value: msg.Value,
				Headers: append(msg.Headers,
					kafka.Header{Key: "x-exception-message", Value: []byte(err.Error())},
					kafka.Header{Key: "x-original-topic", Value: []byte(msg.Topic)},
				),
			})
		}
		r.CommitMessages(ctx, msg)
	}
}

func processWithRetry(ctx context.Context, msg kafka.Message, attempts int, delay time.Duration) error {
	var err error
	for i := 0; i < attempts; i++ {
		if err = process(ctx, msg); err == nil {
			return nil
		}
		time.Sleep(delay)
	}
	return err
}

Здесь: 3 попытки с интервалом 1 секунда, после третьей неудачи — сообщение уходит в orders.DLT (топик, на который настроен dlq-Writer). В DLT-сообщение добавляются заголовки с текстом ошибки и именем оригинального топика.

Retry-топики с нарастающей задержкой

Повторы на месте блокируют партицию на время attempts × delay. Чтобы не блокировать, сообщения перекладывают в отдельные retry-топики — orders-retry-0, orders-retry-1, orders-retry-2 — с нарастающей задержкой, а после последнего — в orders-dlt. Потребитель retry-топика выдерживает паузу, ориентируясь на время записи сообщения:

// Потребитель retry-топика: выдерживаем задержку перед новой попыткой
msg, err := retryReader.FetchMessage(ctx)
if err != nil {
	return
}
if wait := time.Until(msg.Time.Add(30 * time.Second)); wait > 0 {
	time.Sleep(wait)
}
if err := process(ctx, msg); err != nil {
	nextWriter.WriteMessages(ctx, kafka.Message{Key: msg.Key, Value: msg.Value, Headers: msg.Headers})
}
retryReader.CommitMessages(ctx, msg)

Сообщения «отстаиваются» в retry-топиках, не блокируя основной поток.

Когда что использовать

• Временная проблема (база недоступна, внешний сервис не отвечает) → retry-топики с нарастающей задержкой.
• Постоянная проблема (невалидное сообщение, неверный формат) → без повторов, сразу в DLQ.
• Нужно различать типы ошибок → проверка через errors.Is / errors.As: для «невалидных» ошибок сразу DLQ, для остальных — повторы.

Schema Registry: как не сломать соседей при изменении схемы

Когда продьюсер и потребитель — разные сервисы разных команд, любое изменение структуры события потенциально ломает потребитель. Schema Registry решает это централизованным хранилищем схем с проверкой совместимости.

Как это работает:

1. Продьюсер регистрирует схему в Schema Registry и получает числовой идентификатор (schema_id).
2. В каждое сообщение он пишет 4 байта с этим идентификатором, а потом — сжатый payload.
3. Потребитель читает schema_id, один раз скачивает схему из Schema Registry (потом кэширует) и десериализует payload.

Avro, Protobuf или JSON Schema

Стандарт в Kafka-экосистеме — Avro: компактный бинарный формат с хорошей поддержкой эволюции схем; в Go типы генерируют через hamba/avro, а с Schema Registry работают через клиент riferrei/srclient. Protobuf берут команды с gRPC-инфраструктурой. JSON Schema читаем глазами, но занимает в разы больше места — подходит для отладки.

Режимы совместимости

Schema Registry проверяет, не сломает ли новая версия схемы уже работающих потребителей:

• BACKWARD (по умолчанию) — новый потребитель может читать старые сообщения. Можно удалять необязательные поля и добавлять необязательные с default-значением. Нельзя добавлять обязательные поля.
• FORWARD — старый потребитель может читать новые сообщения. Зеркало BACKWARD.
• FULL — оба режима одновременно. Самый строгий.
• NONE — без проверок. Только для особых случаев.

Практическое правило: BACKWARD для топиков, где потребителей много, а продьюсер один — это типичная событийная шина.

Consumer lag: почему потребитель отстаёт

Lag — разница между последним сообщением в партиции и тем, до которого дошёл потребитель. Если lag растёт — потребитель не успевает за продьюсером. Это главная метрика здоровья потребительа в production.

Типичные причины:

1. Медленная обработка — каждое сообщение делает синхронный запрос к базе или внешнему сервису. Решение: распараллелить обработку по goroutine, перейти на пакетные запросы.
2. Мало партиций на группу — запустить больше goroutine с Reader-ами или добавить экземпляры сервиса (но не больше числа партиций).
3. Истекает SessionTimeout — если процесс завис и не шлёт heartbeat дольше SessionTimeout (30 секунд по умолчанию в kafka-go), Kafka считает потребительа мёртвым и запускает перераспределение партиций.
4. Паузы GC или троттлинг CPU в контейнере — процесс встал на десятки секунд, lag вырос на тысячи сообщений.

Посмотреть lag вручную:

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
    --group billing-service --describe
# TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# orders  0          124530          124530          0
# orders  1          124450          124530          80   ← отстаёт

Метрика для мониторинга: kafka_consumergroup_lag. Алерт: lag больше N и растёт N минут подряд.

Тюнинг производительности

Kafka работает хорошо из коробки, но несколько параметров стоит знать.

Продьюсер (Writer)

Потребитель (Reader)

Лучший алгоритм сжатия для большинства случаев — zstd (Kafka 2.1+): сжимает почти как gzip, работает почти как lz4.

Безопасность: SASL, SSL, ACL

По умолчанию Kafka слушает открытый порт без аутентификации — в production так не оставляют. Три слоя защиты:

• SSL/TLS — шифрование соединения. Сертификаты на брокерах и клиентах.
• SASL — аутентификация. SASL/PLAIN — логин и пароль (только поверх SSL). SASL/SCRAM-SHA-256 — безопаснее: challenge-response, пароль по сети не передаётся. SASL/OAUTHBEARER — OAuth2-токены, для интеграции с провайдером идентификации.
• ACL — авторизация: кому разрешено читать и писать в какой топик. Управляется через kafka-acls.sh.

Типичная схема: один SASL-пользователь на каждый сервис, ACL ограничивают его доступ строго до нужных топиков.

mechanism, err := scram.Mechanism(scram.SHA256, "billing-service", os.Getenv("KAFKA_PASSWORD"))
if err != nil {
	log.Fatal(err)
}

dialer := &kafka.Dialer{
	SASLMechanism: mechanism,
	TLS:           &tls.Config{},
}

reader := kafka.NewReader(kafka.ReaderConfig{
	Brokers: []string{"broker:9093"},
	Topic:   "orders",
	GroupID: "billing-service",
	Dialer:  dialer,
})

KRaft: Kafka без ZooKeeper

Раньше Kafka требовала отдельного ZooKeeper-кластера для хранения метаданных: какие топики существуют, кто лидер партиции, какие ACL действуют. ZooKeeper — отдельная система, которую нужно отдельно поднимать, мониторить и чинить.

С версии 3.3 (2022) появился KRaft — собственный протокол на базе Raft внутри самих брокеров. ZooKeeper больше не нужен. В Kafka 4.x поддержка ZooKeeper убрана полностью — все новые кластеры работают только на KRaft.

Для разработчика это почти незаметно: bootstrap-servers и поведение клиента не изменились. Меняется только то, как ops-команда разворачивает и обслуживает кластер.

Коротко

• Writer — отправка, Reader — приём. Параллелизм — несколько goroutine с Reader-ами в одной группе.
• Момент коммита оффсета: ReadMessage — сразу при чтении, CommitInterval — пачками в фоне, FetchMessage + CommitMessages — вручную после обработки.
• Заголовки сообщений — место для технической метаинформации (trace-id, версия схемы). Бизнес-данные — в payload.
• DLQ спасает от зависания потребительа на проблемном сообщении. Готовой обвязки в kafka-go нет — retry-топики и DLQ собираются вручную из Reader и Writer.
• Schema Registry хранит схемы и проверяет совместимость при изменениях. Режим BACKWARD — новый потребитель читает старые сообщения.
• Consumer lag — главная метрика здоровья потребительа. Растёт → потребитель не успевает.
• Тюнинг Writer: BatchTimeout, BatchBytes, Compression: kafka.Zstd. Тюнинг Reader: MinBytes, QueueCapacity.
• Безопасность: SSL для шифрования, SASL для аутентификации, ACL для авторизации по топикам.
• KRaft: с Kafka 3.3+ ZooKeeper не нужен; в Kafka 4.x поддержка ZooKeeper убрана полностью.

Что почитать дальше

• Основы Kafka — устройство брокера, партиции, гарантии доставки, retention.
• Распределённые паттерны — Saga, Outbox, Idempotent Consumer.
• Resilience-паттерны — Circuit Breaker и Timeout в контексте событийных систем.