Kafka в production на TypeScript: kafkajs, DLQ, Schema Registry, тюнинг, безопасность

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

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

kafkajs: producer и eachMessage

Библиотека kafkajs — стандартный Kafka-клиент для Node.js, реализующий протокол Kafka на чистом JavaScript и дающий удобный API. Для отправки сообщений используют producer.send, для приёма — consumer.run с обработчиком eachMessage.

// Отправка
@Injectable()
export class OrderEventPublisher {
  constructor(private readonly producer: Producer) {}

  async publish(event: OrderEvent): Promise<void> {
    await this.producer.send({
      topic: 'orders',
      messages: [{ key: event.orderId.toString(), value: JSON.stringify(event) }],
    });
  }
}

// Приём
const consumer = kafka.consumer({ groupId: 'billing-service' });
await consumer.subscribe({ topic: 'orders' });
await consumer.run({
  eachMessage: async ({ message }) => {
    const event: OrderEvent = JSON.parse(message.value!.toString());
    // обработка
  },
});

По умолчанию consumer.run обрабатывает партиции по очереди, но можно указать больше (partitionsConsumedConcurrently: 3) — тогда несколько партиций обрабатываются параллельно.

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

Kafka следит за тем, до какого сообщения дошёл потребитель, через offset commit. Настройки consumer.run определяют, когда этот коммит происходит:

• autoCommit: true — оффсеты коммитятся автоматически, по интервалу (autoCommitInterval) или по порогу сообщений (autoCommitThreshold) (значение по умолчанию).
• eachMessage + autoCommit — kafkajs отмечает оффсет после каждого успешно обработанного сообщения. Безопаснее при сбоях, но медленнее.
• autoCommit: false — приложение само вызывает consumer.commitOffsets(). Нужен, когда важно закоммитить оффсет только после успешной записи в базу.

await consumer.run({
  partitionsConsumedConcurrently: 3,
  autoCommit: false,
  eachMessage: async ({ topic, partition, message }) => {
    await saveToDatabase(message);
    await consumer.commitOffsets([
      { topic, partition, offset: (Number(message.offset) + 1).toString() },
    ]);
  },
});

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

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

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

// При отправке
await producer.send({
  topic: 'orders',
  messages: [{
    key,
    value: JSON.stringify(event),
    headers: { 'X-Correlation-ID': correlationId },
  }],
});

// При получении
await consumer.run({
  eachMessage: async ({ message }) => {
    const correlationId = message.headers?.['X-Correlation-ID']?.toString();
    await asyncLocalStorage.run({ correlationId }, async () => {
      // обработка
    });
  },
});

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

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

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

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

Простой способ: повторы в обработчике и публикация в DLT

await consumer.run({
  eachMessage: async ({ topic, message }) => {
    try {
      await retry(() => handle(message), { retries: 3, minTimeout: 1000 });
    } catch (e) {
      await producer.send({
        topic: 'orders.DLT',
        messages: [{
          key: message.key,
          value: message.value,
          headers: {
            'x-exception-message': (e as Error).message,
            'x-original-topic': topic,
          },
        }],
      });
    }
  },
});

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

Способ без блокировки: retry-топики

Готового декларативного аналога в kafkajs нет — тот же паттерн собирается вручную: неудачное сообщение переотправляется в retry-топик со всё большей задержкой, а после исчерпания попыток — в DLT.

const RETRY_TOPICS = ['orders-retry-0', 'orders-retry-1', 'orders-retry-2'];

async function reroute(message: KafkaMessage, attempt: number, error: Error): Promise<void> {
  const next = attempt < RETRY_TOPICS.length ? RETRY_TOPICS[attempt] : 'orders-dlt';
  await producer.send({
    topic: next,
    messages: [{
      key: message.key,
      value: message.value,
      headers: {
        'x-attempt': String(attempt + 1),
        'x-retry-at': String(Date.now() + 1000 * 2 ** attempt),
        'x-exception-message': error.message,
      },
    }],
  });
}

Потребитель retry-топиков перед обработкой ждёт наступления x-retry-at. Сообщения «отстаиваются» в топиках orders-retry-0, orders-retry-1, orders-retry-2, не блокируя основной поток.

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

• Временная проблема (база недоступна, внешний сервис не отвечает) → retry-топики с нарастающей задержкой.
• Постоянная проблема (невалидное сообщение, неверный формат) → без повторов, сразу в DLQ.
• Нужно различать типы ошибок → проверять тип исключения в catch: например, SyntaxError из JSON.parse — сразу 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: компактный бинарный формат с хорошей поддержкой эволюции схем; в Node.js с ним работают через @kafkajs/confluent-schema-registry. Protobuf берут команды с gRPC-инфраструктурой. JSON Schema читаем глазами, но занимает в разы больше места — подходит для отладки.

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

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

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

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

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

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

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

1. Медленная обработка — каждое сообщение делает синхронный запрос к базе или внешнему сервису. Решение: уменьшить размер пачки (maxBytesPerPartition), параллелизировать обработку, перейти на пакетные запросы.
2. Мало партиций на группу — увеличить partitionsConsumedConcurrently или добавить экземпляры сервиса (но не больше числа партиций).
3. Истекает sessionTimeout — если обработка занимает дольше таймаута сессии (30 секунд по умолчанию) без heartbeat, Kafka считает потребительа мёртвым и запускает перераспределение партиций. Решение: вызывать heartbeat() во время долгой обработки или поднять sessionTimeout.
4. Блокировка event loop / паузы GC — процесс встал на 30 секунд, 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 работает хорошо из коробки, но несколько параметров стоит знать.

Продьюсер

Потребитель

Лучший алгоритм сжатия для большинства случаев — 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 ограничивают его доступ строго до нужных топиков.

const kafka = new Kafka({
  clientId: 'billing-service',
  brokers: ['kafka:9093'],
  ssl: {
    ca: [fs.readFileSync('/etc/kafka/ca.pem', 'utf-8')],
  },
  sasl: {
    mechanism: 'scram-sha-256',
    username: 'billing-service',
    password: process.env.KAFKA_PASSWORD!,
  },
});

KRaft: Kafka без ZooKeeper

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

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

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

Коротко

• producer.send — отправка, consumer.run с eachMessage — приём. Параллелизм по партициям — partitionsConsumedConcurrently.
• Commit-режим управляет тем, когда коммитится оффсет: autoCommit по интервалу/порогу (по умолчанию) или ручной commitOffsets после записи в базу.
• Заголовки сообщений — место для технической метаинформации (trace-id, версия схемы). Бизнес-данные — в payload.
• DLQ спасает от зависания потребительа на проблемном сообщении. Retry-топики с нарастающей задержкой — способ повторов без блокировки основного потока.
• Schema Registry хранит схемы и проверяет совместимость при изменениях. Режим BACKWARD — новый потребитель читает старые сообщения.
• Consumer lag — главная метрика здоровья потребительа. Растёт → потребитель не успевает.
• Тюнинг producer: пачка сообщений в одном send, compression=zstd. Тюнинг consumer: minBytes, maxBytesPerPartition, sessionTimeout.
• Безопасность: SSL для шифрования, SASL для аутентификации, ACL для авторизации по топикам.
• KRaft: с Kafka 3.3+ ZooKeeper не нужен; в Kafka 4.x поддержка ZooKeeper убрана полностью.

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

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