> Где размещать воркеры, читающие данные из Kafka и отправляющие в аналитику (Go)

Короткий ответ

Воркеры, читающие из Kafka и отправляющие в аналитику, размещаются в отдельном сервисе-консьюмере, который масштабируется горизонтально через consumer group. Ключевое правило - воркер не должен совмещать чтение из Kafka с обработкой HTTP-запросов или другими блокирующими задачами. Оптимально - отдельный deployment (в Kubernetes) или отдельный процесс, с конфигурацией через env-переменные. Это позволяет независимо масштабировать потребление, управлять ребалансировкой и изоляцией сбоев.

Подробное объяснение

Размещение воркеров зависит от архитектуры, но есть общие принципы:

• Отдельный сервис - самый чистый вариант. Воркер - это самостоятельный бинарник, который только читает из Kafka и пишет в аналитику. Он не имеет HTTP-эндпоинтов, не хранит состояние, не зависит от основного API. Это упрощает деплой, мониторинг и масштабирование.
• Внутри основного сервиса - допустимо только если нагрузка минимальна и вы уверены, что обработка сообщений не заблокирует основной поток. В Go это возможно через отдельные goroutine, но тогда сложнее управлять жизненным циклом и ребалансировкой.
• Consumer group - все воркеры, читающие одну тему, должны быть в одной consumer group. Kafka сам распределяет партиции между воркерами. Если воркеров больше, чем партиций, часть будет простаивать - это нормально.
• Изоляция по темам - если воркеры обрабатывают разные типы данных, лучше разделять их на разные consumer group или даже разные сервисы. Это позволяет независимо менять конфигурацию (batch size, retries, DLQ).
• Размещение в инфраструктуре - в Kubernetes это отдельный Deployment с HPA по метрике lag (например, kafka_consumer_lag). Воркеры должны быть stateless, чтобы можно было безопасно добавлять и удалять инстансы.
• Сеть - воркер должен находиться в той же сети, что и Kafka и аналитическая система, чтобы минимизировать задержки. Если аналитика - это ClickHouse или Greenplum, воркер пишет батчами, а не по одному сообщению.

На практике

• Создаёте отдельный модуль или сервис kafka-worker в монорепозитории.
• Используете библиотеку segmentio/kafka-go или confluent-kafka-go (обёртка над librdkafka). Для Go чаще выбирают segmentio/kafka-go - он чище и без CGO.
• Воркер запускается как long-running процесс, который слушает сигналы SIGTERM/SIGINT для graceful shutdown.
• Настройки: group.id, bootstrap.servers, auto.offset.reset (обычно earliest для аналитики, чтобы не терять данные), enable.auto.commit - лучше false, коммитить вручную после успешной отправки в аналитику.
• Для аналитики используете батчинг: накапливаете N сообщений или ждёте T миллисекунд, затем отправляете одним запросом. Это критично для производительности.
• Обязательно настраиваете DLQ (dead letter queue) - отдельная тема для сообщений, которые не удалось обработать после N ретраев.
• Мониторинг: метрики lag, количество обработанных сообщений, ошибки отправки в аналитику, время обработки батча.

Пример кода

GO
package main

import (
    "context"
    "log"
    "os"
    "os/signal"
    "syscall"
    "time"

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

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // graceful shutdown
    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
    go func() {
        <-sigCh
        cancel()
    }()

    reader := kafka.NewReader(kafka.ReaderConfig{
        Brokers:        []string{os.Getenv("KAFKA_BROKERS")},
        GroupID:        "analytics-worker",
        Topic:          "events",
        MinBytes:       10e3, // 10KB
        MaxBytes:       10e6, // 10MB
        CommitInterval: time.Second,
        StartOffset:    kafka.FirstOffset,
    })
    defer reader.Close()

    batch := make([]kafka.Message, 0, 100)
    batchSize := 100
    flushInterval := 5 * time.Second

    flush := func() {
        if len(batch) == 0 {
            return
        }
        // отправка батча в аналитику (ClickHouse, например)
        if err := sendToAnalytics(ctx, batch); err != nil {
            log.Printf("failed to send batch: %v", err)
            // здесь можно отправить в DLQ и закоммитить, или ретраить
            return
        }
        batch = batch[:0]
    }

    ticker := time.NewTicker(flushInterval)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            flush() // финальный flush
            return
        case <-ticker.C:
            flush()
        default:
            msg, err := reader.FetchMessage(ctx)
            if err != nil {
                if ctx.Err() != nil {
                    continue
                }
                log.Printf("fetch error: %v", err)
                time.Sleep(time.Second)
                continue
            }
            batch = append(batch, msg)
            if len(batch) >= batchSize {
                flush()
            }
            // коммитим после успешной отправки
            if err := reader.CommitMessages(ctx, msg); err != nil {
                log.Printf("commit error: %v", err)
            }
        }
    }
}

Как отвечать на собеседовании

Начните с главного - отдельный сервис и consumer group. Объясните, почему это важно: изоляция сбоев, независимое масштабирование, простота мониторинга. Затем переходите к деталям: батчинг, ручной коммит, DLQ. Если спросят про Kubernetes - упомяните HPA по lag. Если спросят про альтернативы - скажите, что можно встроить в основной сервис, но это оправдано только при низкой нагрузке. Покажите понимание trade-off: отдельный сервис - больше инфраструктуры, но лучше управляемость.

Что проверяет интервьюер

• Понимание модели consumer group и партиционирования в Kafka.
• Умение проектировать отказоустойчивые консьюмеры: ручной коммит, retries, DLQ.
• Знание практик масштабирования: stateless-воркеры, батчинг, горизонтальное масштабирование.
• Понимание graceful shutdown и обработки сигналов в Go.
• Умение обосновать архитектурное решение, а не просто назвать паттерн.

Типичные ошибки

• Размещение воркера внутри HTTP-сервиса без изоляции - при рестарте или ребалансировке теряются сообщения.
• Автокоммит оффсетов - при сбое после чтения, но до отправки в аналитику, данные теряются.
• Отправка каждого сообщения отдельно в аналитику - убивает производительность.
• Игнорирование ребалансировки - при добавлении нового воркера старые могут получить ошибки, если не обработать Rebalance callback.
• Нет DLQ - сообщение с битыми данными блокирует всю партицию.
• Нет мониторинга lag - проблема обнаруживается через часы после начала.