Go-клиент Kafka: продюсер и консьюмер с изоляцией по партициям, OpenTelemetry и graceful shutdown из коробки. Построен на franz-go.

Чистый Go, без cgo: статическая сборка, обычная кросс-компиляция, FROM scratch и Alpine работают без тегов сборки, детектор гонок видит весь код клиента.

go get github.com/alfzs/kafkax/v2

import "github.com/alfzs/kafkax/v2"

Требуется Go 1.26+.

cfg := kafkax.Config{
    Brokers:         []string{"kafka:9092"},
    ClientID:        "my-service",
    GracefulTimeout: 3 * time.Minute,
    DialTimeout:     10 * time.Second,
    Producer: kafkax.Producer{
        RequiredAcks:      -1,
        EnableIdempotence: true,
        MessageTimeout:    30 * time.Second,
        FlushTimeout:      time.Minute,
    },
}

producer, err := kafkax.NewKafkaProducer(cfg)
if err != nil {
    log.Fatal(err)
}
defer producer.Close()

err = producer.SendMessage(ctx, kafkax.PublishRequest{
    Topic: "orders",
    Key:   orderID[:],
    Value: payload,
    Headers: kafkax.Headers{
        {Key: "signature", Value: signature},
    },
})

SendMessage синхронен и потокобезопасен: параллельные вызовы батчатся внутри franz-go, так что «синхронный» не означает «по запросу на сообщение». Весь путь отправки укладывается в один бюджет — Producer.MessageTimeout, отсчитываемый от входа в метод.

Ключи traceparent, tracestate и baggage зарезервированы за W3C trace propagation: SendMessage вернёт ErrReservedHeaderKey, если один из них встретится в PublishRequest.Headers.

type orderHandler struct{}

func (h *orderHandler) ProcessMessage(ctx context.Context, msg kafkax.IncomingMessage) error {
    // ctx содержит OTel-span — используйте его для дочерних операций
    if sig, ok := msg.Headers.Get("signature"); ok {
        // проверка подписи
    }
    return h.store(ctx, msg.Value)
}

cfg.Consumer = kafkax.Consumer{
    Group:             "my-service.group",
    InitialOffset:     kafkax.OffsetEarliest,
    SessionTimeout:    45 * time.Second,
    HeartbeatInterval: 3 * time.Second,
    RebalanceTimeout:  time.Minute,
    CommitInterval:    5 * time.Second,
    HandlerMaxRetries: 3,
    HandlerRetryDelay: time.Second,
}

consumer, err := kafkax.NewKafkaConsumer(cfg)
if err != nil {
    log.Fatal(err)
}

if err := consumer.AddHandler("orders", &orderHandler{}); err != nil {
    log.Fatal(err)
}

if err := consumer.Start(ctx); err != nil { // не блокирует
    log.Fatal(err)
}
defer consumer.Stop()

Обработчики регистрируются до Start; подписка на топики происходит внутри Start, отдельного шага для неё нет. Повторный Start вернёт ErrConsumerStarted, консьюмер после Stop не перезапускается.

Буферы записей. IncomingMessage.Key, .Value и .Headers ссылаются на буферы franz-go и валидны только на время вызова ProcessMessage. Если данные нужны дольше — копируйте.

At-least-once. Оффсет отмечается к коммиту только после того, как обработчик вернул nil, а фоновый коммит (Consumer.CommitInterval) отправляет брокеру исключительно отмеченное. Автокоммит по факту чтения не используется: он двигал бы оффсет до обработки и превращал гарантию в at-most-once при падении процесса.

Следствие, о котором нужно знать при внедрении: дубликаты штатны. Сообщение, обработанное непосредственно перед падением или ребалансом, будет обработано повторно. Обработчик обязан быть идемпотентным — exactly-once библиотека не даёт и дать не может.

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

Когда ProcessMessage возвращает ошибку, сообщение проходит два независимых этапа.

Управляется Consumer.HandlerMaxRetries и Consumer.HandlerRetryDelay:

Повторы идут в горутине партиционного воркера и блокируют партицию: пока сообщение повторяется, следующие сообщения этой же партиции ждут. Это цена сохранения порядка, а не недосмотр. При -1 партиция заблокирована, пока обработчик не вернёт nil или консьюмер не остановят; этап 2 не наступает никогда.

Паника внутри ProcessMessage перехватывается, оборачивается в ErrHandlerPanic и идёт тем же путём, что обычная ошибка, — воркер не падает.

Наступает, когда повторы исчерпаны. Исходов два, и выбирает между ними наличие Config.OnMessageSkipped:

Хук задан и вернул nil — сообщение считается пропущенным: оффсет отмечается, коммит двигается дальше, партиция продолжает работу. Метрика — status="skipped".

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

cfg.OnMessageSkipped = func(ctx context.Context, msg kafkax.IncomingMessage, cause error) error {
    return dlq.Publish(ctx, msg, cause) // nil ⇒ оффсет двигается
}

Хук не задан, вернул ошибку или запаниковал — оффсет НЕ отмечается, партиция ставится на паузу и остаётся на непрокоммиченном оффсете. Метрика — status="error", в лог пишется запись уровня Error. Сообщение приедет снова после ребаланса или перезапуска процесса. Остальные партиции продолжают работать.

Застрявшая партиция видна сразу — по растущему лагу, по kafkax.consumer.messages.processed{status="error"} и по логу. Потерянное сообщение не видно ничем. Пропуск — явное действие потребителя пакета, а не то, что случается само.

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

Важно. Пока партиция остаётся за тем же экземпляром и ребаланса не происходит, отравленное сообщение не приезжает заново само по себе. В однопроцессном развёртывании это значит: до перезапуска (или до входа в группу второго экземпляра) партиция стоит. Признак — растущий лаг без новых записей уровня Error: сообщение до обработчика больше не доходит, поэтому и счётчик status="error" перестаёт расти.

Отдельный случай: если контекст отменён во время паузы между повторами (идёт остановка консьюмера), сообщение не отмечается и партицию не травит — оно просто приедет снова.

Sentinel-ошибки делятся по одному признаку — можно ли повторить, не рискуя дубликатом:

Проверяются через errors.Is; текст сообщения частью контракта не является.

AddHandler принимает опциональную цепочку ConsumerMiddleware — обёртку над ConsumerHandler в духе http.Handler:

type ConsumerMiddleware func(ConsumerHandler) ConsumerHandler

Middleware применяются в порядке перечисления: первый в списке — внешний, выполняется первым и может решить не звать next вовсе.

consumer.AddHandler("orders", &orderHandler{}, loggingMiddleware, metricsMiddleware)

Функцию можно передать как обработчик через ConsumerHandlerFunc, а собрать цепочку вне консьюмера — через kafkax.Chain.

Готовая middleware для маршрутизации сообщений с композитным ключом: не тот адресат тихо пропускается, до ProcessMessage дело не доходит.

consumer.AddHandler("events", &orderHandler{},
    kafkax.MatchKeyMiddleware(myTenantID, myExternalBotID))

encoding.EncodeKey собирает бинарный ключ Kafka-сообщения из нескольких значений (uuid.UUID, string, int64, bool) — без обратного декодирования: консьюмер знает свои значения и сравнивает, а не разбирает чужой ключ.

import "github.com/alfzs/kafkax/v2/encoding"

// Продюсер
key, err := encoding.EncodeKey(tenantID, externalBotID)
if err != nil {
    return err
}

producer.SendMessage(ctx, kafkax.PublishRequest{Topic: "events", Key: key, Value: payload})

// Консьюмер — вручную, без middleware
func (h *handler) ProcessMessage(ctx context.Context, msg kafkax.IncomingMessage) error {
    if !encoding.MatchKey(msg.Key, myTenantID, myExternalBotID) {
        return nil // не наш адресат
    }
    ...
}

encoding.ValidateKeyLength(key, parts...) проверяет, что key не короче длины, которую дал бы EncodeKey(parts...), и возвращает ErrInvalidKey, если это не так, — сигнал усечённого или повреждённого сообщения, в отличие от валидного по длине ключа другого тенанта (для него MatchKey просто вернёт false). MatchKeyMiddleware делает эту проверку сама.

Цикл опроса раздаёт записи партиционным воркерам: по горутине на назначенную партицию, порядок внутри партиции сохраняется, разные партиции обрабатываются параллельно.

poll ──┬─► partition 0 ──► ProcessMessage (последовательно)
       ├─► partition 1 ──► ProcessMessage (последовательно)
       └─► partition 2 ──► ProcessMessage (последовательно)

Пропускная способность настраивается числом партиций топика, а не параметрами библиотеки. Consumer.MessageQueueSize задаёт, насколько цикл опроса может обгонять обработку.

У продюсера собственного слоя очередей и воркеров нет: батчинг, упорядочивание по партиции и лимит памяти делает клиент Kafka. Backpressure — это Producer.MaxBufferedRecords / MaxBufferedBytes: при заполнении буфера SendMessage ждёт освобождения места.

Трассировка и транспортные метрики приходят из kotel по семантическим соглашениям OTel: контекст переносится через заголовки, спаны именуются по конвенции.

[HTTP handler span]
    └── [Producer span]  ← SendMessage
            └── [Consumer span]  ← ProcessMessage (ctx содержит parent span)
                    └── [DB query span]

Используются глобальные провайдеры otel.GetTracerProvider() и otel.GetMeterProvider(). OTel не обязателен — при ненастроенных провайдерах работают no-op реализации: спаны и метрики отбрасываются, в заголовки ничего не пишется. Функциональность пакета от этого не зависит.

Длительности — в секундах, как требует OTel. Атрибута partition нет ни у одной метрики: он умножает кардинальность на число партиций, не давая ничего сверх того, что уже есть в спане.

Транспортные метрики (соединения, байты, ошибки чтения и записи) регистрирует kotel под своими именами.

Отдельного поля с протоколом безопасности нет: протокол выводится из конфигурации. TLS включается флагом TLS.Enabled, SASL — непустым SASL.Mechanism.

Программные поля (без env и yaml):

Применяется только при непустом Mechanism.

Config.Validate() проверяет обе секции сразу — для приложения, создающего из одного Config и продюсер, и консьюмер. Конструкторы вызывают проверку только своей роли: продюсеру незачем требовать consumer.group, консьюмеру — producer.flush_timeout. Поэтому Config, прошедший NewKafkaProducer, может не пройти Validate.

Ошибки собираются все разом через errors.Join, а не возвращаются по первой: иначе неполный конфиг чинится по одному полю за перезапуск. Список разворачивается через errors.Unwrap() []error.

Теги env совместимы с cleanenv:

var cfg kafkax.Config
if err := cleanenv.ReadEnv(&cfg); err != nil {
    log.Fatal(err)
}

KAFKAX_BROKERS=kafka-1.example.com:9092,kafka-2.example.com:9092
KAFKAX_CLIENT_ID=my-service
KAFKAX_CONSUMER_GROUP=my-service.group
KAFKAX_TLS_ENABLED=true
KAFKAX_SASL_MECHANISM=SCRAM-SHA-512
KAFKAX_SASL_USERNAME=my-service-user
KAFKAX_SASL_PASSWORD=secret

Продюсер. Close() перестаёт принимать новые отправки, дожидается завершения тех, что уже в полёте, и досылает буферизованное финальным flush. GracefulTimeout — общий бюджет на обе фазы, FlushTimeout ограничивает сверху только вторую. Повторный вызов безопасен.

Консьюмер. Stop() останавливает цикл опроса, дожидается партиционных воркеров в пределах GracefulTimeout, коммитит отмеченное и покидает группу. Повторный вызов безопасен.

Оба клиента реагируют на отмену контекста, переданного в Start, но явные Close/Stop предпочтительнее: только они гарантируют фазу дообработки.

MessageProducer и MessageConsumer объявлены в пакете, чтобы вызывающий код подменял клиента в тестах, не поднимая брокер:

type MessageProducer interface {
    SendMessage(ctx context.Context, req PublishRequest) error
    Close() error
}