@packages/kafka-client

Полнофункциональный Kafka клиент для NestJS с поддержкой паттернов Fire-and-Forget, Request-Reply, Retry механизмом и Dead Letter Queue (DLQ).

📋 Содержание

• Возможности
• Требования
• Установка
• Развертывание в Docker
• Структура пакета
• Быстрый старт
• Использование модулей и сервисов
• API Reference
• Типы и интерфейсы
• Troubleshooting
• Тестирование

🚀 Возможности

• ✅ Fire-and-Forget паттерн - отправка сообщений без ожидания ответа
• ✅ Request-Reply паттерн - отправка запросов с ожиданием ответа через correlation ID
• ✅ Retry механизм - автоматические повторные попытки с exponential backoff
• ✅ Dead Letter Queue (DLQ) - обработка сообщений, которые не удалось обработать
• ✅ NestJS модули - готовые глобальные модули для простой интеграции
• ✅ Низкоуровневый API - прямой доступ к KafkaCore для сложных сценариев
• ✅ 100% покрытие тестами - надежность и качество кода
• ✅ TypeScript типизация - полная типобезопасность
• ✅ Graceful shutdown - корректное отключение при остановке приложения

📋 Требования

• Node.js: >= 22.11.0
• pnpm: >= 10.18.0
• NestJS: >= 11.0.0
• Kafka: >= 2.0.0 (через kafkajs)

📦 Установка

npm install @packages/kafka-client

Зависимости

Пакет требует следующие peer dependencies:

{
  "@nestjs/common": "^11.0.0",
  "@nestjs/microservices": "^11.0.0",
  "kafkajs": "^2.0.0",
  "reflect-metadata": "^0.1.13 || ^0.2.0",
  "rxjs": "^7.0.0"
}

🐳 Развертывание в Docker

Dockerfile

Пакет включает готовый Dockerfile для сборки образа:

FROM node:22-alpine AS base
RUN corepack enable && corepack prepare [email protected] --activate
WORKDIR /app
COPY package.json pnpm-lock.yaml* ./
RUN pnpm install --frozen-lockfile
COPY . .
RUN pnpm run build

FROM node:22-alpine AS production
WORKDIR /app
COPY package.json pnpm-lock.yaml* ./
RUN corepack enable && corepack prepare [email protected] --activate && \
    pnpm install --frozen-lockfile --prod
COPY --from=base /app/dist ./dist
RUN addgroup -g 1001 -S nodejs && \
    adduser -S nodejs -u 1001 && \
    chown -R nodejs:nodejs /app
USER nodejs
CMD ["node", "dist/index.js"]

Сборка образа

docker build -t kafka-client:latest .

Использование в docker-compose.yml

services:
  kafka:
    image: confluentinc/cp-kafka:latest
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    ports:
      - "9092:9092"

  test-service:
    build: .
    environment:
      KAFKA_BROKERS: kafka:9092
      KAFKA_CLIENT_ID: test-service
    depends_on:
      - kafka

📁 Структура пакета

src/
├── core/                                 # Низкоуровневый API
│   ├── kafka.core.ts                     # Основной класс KafkaCore
│   ├── patterns/                         # Паттерны отправки сообщений
│   │   ├── fire-and-forget.pattern.ts    # Fire-and-Forget паттерн
│   │   └── request-reply.pattern.ts      # Request-Reply паттерн
│   └── retry/                            # Retry и DLQ механизмы
│       ├── retry-handler.ts              # Обработка retry
│       └── dlq-handler.ts                # Обработка DLQ
│
├── main/                                 # NestJS модули (РЕКОМЕНДУЕТСЯ)
│   ├── client/                           # Базовый модуль подключения
│   │   ├── kafka-client.module.ts        # KafkaClientModule
│   │   └── kafka-client.service.ts       # KafkaClientService
│   ├── producer/                         # Producer модуль
│   │   ├── kafka-producer.module.ts      # KafkaProducerModule
│   │   └── kafka-producer.service.ts     # KafkaProducerService
│   └── consumer/                         # Consumer модуль
│       ├── kafka-consumer.module.ts      # KafkaConsumerModule
│       └── kafka-consumer.service.ts     # KafkaConsumerService
│
├── types/                                # TypeScript типы и интерфейсы
│   ├── kafka-topics.ts                   # Enum топиков и конфигурация
│   ├── kafka-message.ts                  # Типы сообщений
│   ├── module-options.interface.ts       # Опции модулей
│   ├── kafka-core.options.interface.ts   # Опции KafkaCore
│   ├── request-reply-options.interface.ts # Опции Request-Reply
│   ├── retry-handler.interface.ts        # Опции Retry
│   └── dlq-handler.interface.ts          # Опции DLQ
│
├── errors/                               # Кастомные ошибки
│   ├── kafka-client.error.ts             # KafkaClientError
│   └── rpc-request.error.ts              # RpcRequestError
│
├── utils/                                # Утилиты
│   └── injection-keys.ts                 # Ключи для DI
│
└── index.ts                              # Точка входа (экспорты)

🏗️ Архитектура

Пакет предоставляет два способа использования:

1. NestJS модули (РЕКОМЕНДУЕТСЯ)

Готовые глобальные модули KafkaClientModule, KafkaProducerModule и KafkaConsumerModule для простой интеграции в NestJS приложения.

Преимущества:

• Не нужно дублировать код в каждом сервисе
• Автоматическая инициализация и подключение
• Встроенная интеграция с NestJS DI
• Graceful shutdown
• Единое подключение к Kafka для всего приложения

2. Низкоуровневый API

Прямое использование KafkaCore для полного контроля над подключением и процессами.

🔧 Быстрый старт (NestJS модули)

Шаг 1: Настройка KafkaClientModule (базовый модуль)

ВАЖНО: KafkaClientModule должен быть импортирован ПЕРЕД KafkaProducerModule и KafkaConsumerModule.

// app.module.ts
import { Module } from '@nestjs/common';
import { ConfigModule } from '@nestjs/config';
import { KafkaClientModule, KafkaTopic } from '@packages/kafka-client';

@Module({
  imports: [
    ConfigModule.forRoot({
      isGlobal: true,
      load: [kafkaConfig], // Ваша конфигурация Kafka
    }),
    // Базовый модуль для подключения к Kafka
    KafkaClientModule.forRootAsync<[KafkaConfiguration]>({
      useFactory: (config: KafkaConfiguration) => ({
        brokers: config.brokers,
        clientId: config.clientId,
        responseTopics: [KafkaTopic.TEST_RESPONSES], // Для Request-Reply
        defaultTimeout: 30000, // Таймаут по умолчанию
        connectionTimeout: 10000, // Таймаут подключения (опционально)
        requestTimeout: 30000, // Таймаут запроса (опционально)
        retry: { // Настройки retry для подключения (опционально)
          retries: 8,
          initialRetryTime: 300,
          maxRetryTime: 30000,
        },
      }),
      inject: [kafkaConfig.KEY],
    }),
  ],
})
export class AppModule {}

Шаг 2: Producer (отправка команд)

Для отправляющего собщения сервера (отправляет команды и ждет ответы):

// app.module.ts
import { KafkaProducerModule } from '@packages/kafka-client';

@Module({
  imports: [
    KafkaClientModule.forRootAsync(...), // Из шага 1
    KafkaProducerModule.forRootAsync(), // Producer модуль
  ],
})
export class AppModule {}

Шаг 3: Consumer (обработка команд)

Для принимающего сообщения сервера (принимает команды и отправляет ответы):

// app.module.ts
import { KafkaConsumerModule, KafkaTopic } from '@packages/kafka-client';
import { TestModule } from './test/test.module';
import { TestHandlerService } from './test-message.handler';

@Module({
  imports: [
    KafkaClientModule.forRootAsync(...), // Из шага 1
    TestModule, // Модуль с зависимостями handler'а
    KafkaConsumerModule.forRoot({
      topics: [KafkaTopic.TEST_COMMANDS],
      groupId: 'test-consumer',
      messageHandler: TestHandlerService,
      imports: [TestModule], // Дополнительные модули для DI
    }),
  ],
})
export class AppModule {}

// test-message.handler.ts
import { Injectable } from '@nestjs/common';
import { KafkaMessageHandler, TestCommandType } from '@packages/kafka-client';
import { TestService } from './test.service';
import type { TestCommand } from '@packages/kafka-client';
import { RpcException } from '@nestjs/microservices';
import { HttpStatus } from '@packages/types';

@Injectable()
export class TestHandlerService implements KafkaMessageHandler {
  constructor(private readonly testService: TestService) {}

  async handleMessage(
    topic: string, 
    message: unknown, 
    headers?: Record<string, string>
  ): Promise<unknown> {
    const command = message as TestCommand;

    // Используйте headers для логирования или трейсинга
    const correlationId = headers?.['correlation-id'];
    if (correlationId) {
      this.logger.log(`Processing command with correlation-id: ${correlationId}`);
    }

    switch (command.type) {
      case TestCommandType.CONNECT:
        return await this.testService.connect({
          userId: command.userId,
          apiKey: command.payload.apiKey,
          apiSecret: command.payload.apiSecret,
        });

      case TestCommandType.GET_BALANCE:
        return await this.testService.getBalance(command.payload.accountType);

      default:
        throw new RpcException({
          statusCode: HttpStatus.BAD_REQUEST,
          message: `Unknown command type: ${command.type}`,
          error: 'ValidationError',
        });
    }
  }
}

Готово! Модуль автоматически:

• Подключится к Kafka при старте
• Начнет слушать топики
• Обработает сообщения через handler
• Отправит ответы обратно (для Request-Reply)
• Отключится при shutdown

📚 Использование модулей и сервисов

KafkaClientModule

Назначение: Единое подключение к Kafka для всего приложения.

Метод инициализации:

forRootAsync(options)

KafkaClientModule.forRootAsync<[KafkaConfiguration]>({
  useFactory: (config: KafkaConfiguration) => ({
    brokers: config.brokers,
    clientId: config.clientId,
    responseTopics: config.responseTopics,
    defaultTimeout: config.defaultTimeout,
    connectionTimeout: config.connectionTimeout,
    requestTimeout: config.requestTimeout,
    retry: config.retry,
  }),
  inject: [kafkaConfig.KEY],
  imports: [ConfigModule],
})

Параметры:

• useFactory: (deps) => KafkaClientModuleOptions - фабрика для создания опций модуля
• inject?: InjectionToken[] - зависимости для инжекции в useFactory
• imports?: Module[] - дополнительные модули для DI

Экспортирует: KafkaClientService (для внутреннего использования)

ВАЖНО: Этот модуль должен быть импортирован ПЕРЕД KafkaProducerModule и KafkaConsumerModule.

KafkaProducerModule

Назначение: Модуль для отправки сообщений (Fire-and-Forget и Request-Reply).

ВАЖНО: Требует импорта KafkaClientModule.forRootAsync() перед собой.

forRootAsync(options?)

KafkaProducerModule.forRootAsync({
  imports?: [ConfigModule], // Опционально: дополнительные модули для DI
})

Параметры:

• imports?: Module[] - дополнительные модули для DI (опционально)

Экспортирует: KafkaProducerService

KafkaProducerService

Методы:

sendCommand<TRequest, TResponse>(commandTopic, responseTopic, message, timeout?, additionalHeaders?)

Request-Reply паттерн - отправка команды с ожиданием ответа.

Параметры:

• commandTopic: KafkaTopic - топик для отправки команды
• responseTopic: KafkaTopic - топик для получения ответа
• message: TRequest - сообщение для отправки
• timeout?: number - таймаут ожидания ответа в миллисекундах (опционально, по умолчанию 30000)
• additionalHeaders?: Record<string, string> - дополнительные заголовки для сообщения (опционально)

Возвращает: Promise<TResponse>

Пример:

const response = await kafkaProducer.sendCommand(
  KafkaTopic.TEST_COMMANDS,
  KafkaTopic.TEST_RESPONSES,
  command,
  30000, // timeout
  { 'user-id': '123' } // дополнительные заголовки
);

sendFireAndForget<T>(topic, message)

Fire-and-Forget паттерн - отправка сообщения без ожидания ответа.

Параметры:

• topic: KafkaTopic - топик для отправки
• message: T - сообщение для отправки

Возвращает: Promise<void>

Пример:

await kafkaProducer.sendFireAndForget(
  KafkaTopic.TEST_EVENTS,
  { userId: '123', data: { ... } }
);

isConnected()

Проверка статуса подключения к Kafka.

Возвращает: boolean

Пример:

const isConnected = kafkaProducer.isConnected();
console.log(`Kafka connected: ${isConnected}`);

KafkaConsumerModule

Назначение: Модуль для получения и обработки сообщений.

ВАЖНО: Требует импорта KafkaClientModule.forRootAsync() перед собой.

forRoot(options)

KafkaConsumerModule.forRoot({
  topics: [KafkaTopic.TEST_COMMANDS],
  groupId: 'test-consumer',
  messageHandler: TestHandlerService,
  imports?: [TestModule], // Дополнительные модули для DI
})

Параметры:

• topics: KafkaTopic[] - массив топиков для подписки
• groupId: string - уникальный ID consumer group
• messageHandler: KafkaMessageHandler - класс handler'а, реализующий интерфейс KafkaMessageHandler
• imports?: Module[] - дополнительные модули для DI зависимостей handler'а

Экспортирует: KafkaConsumerService, messageHandler (ваш handler)

KafkaMessageHandler (интерфейс)

Ваш handler должен реализовать этот интерфейс:

interface KafkaMessageHandler {
  handleMessage(
    topic: string, 
    message: unknown, 
    headers?: Record<string, string>
  ): Promise<unknown>;
}

Параметры:

• topic: string - название топика, из которого пришло сообщение
• message: unknown - распарсенное сообщение (JSON объект)
• headers?: Record<string, string> - заголовки сообщения (опционально)

Возвращаемое значение:

• Если возвращаете объект → отправится как ответ (Request-Reply)
• Если возвращаете undefined → ответ не отправляется (Fire-and-Forget)
• Если выбрасываете RpcException → отправится ответ с ошибкой

Пример использования headers:

async handleMessage(topic: string, message: unknown, headers?: Record<string, string>): Promise<unknown> {
  const correlationId = headers?.['correlation-id'];
  const replyTo = headers?.['reply-to'];
  const userId = headers?.['user-id'];
  
  // Используйте headers для логирования, трейсинга и т.д.
  if (correlationId) {
    this.logger.log(`Processing request with correlation-id: ${correlationId}`);
  }
  
  // ... обработка сообщения
}

Обработка ошибок:

import { RpcException } from '@nestjs/microservices';
import { HttpStatus } from '@packages/types';

throw new RpcException({
  statusCode: HttpStatus.BAD_REQUEST,
  message: 'Invalid API key',
  error: 'ValidationError',
});

🔧 Настройка переменных окружения

Для обоих сервисов добавьте в .env:

KAFKA_BROKERS=localhost:9093
KAFKA_CLIENT_ID=test-service

Валидация .env переменных:

# Адреса Kafka брокеров (через запятую для нескольких брокеров)
KAFKA_BROKERS=localhost:9093
# или для Docker: kafka:9092
# или для кластера: broker1:9092,broker2:9092,broker3:9092

# Уникальный идентификатор клиента для вашего сервиса
KAFKA_CLIENT_ID=test-service

🎯 Типы топиков

enum KafkaTopic {
  // Test топики
  TEST_COMMANDS = 'test-commands',
  TEST_RESPONSES = 'test-responses',
  TEST_DLQ = 'test-dlq',
}

Конфигурация топиков

Каждый топик имеет конфигурацию с настройками партиций и retention:

const KAFKA_TOPIC_CONFIG: Record<KafkaTopic, TopicConfig> = {
  [KafkaTopic.TEST_COMMANDS]: {
    partitions: 3,
    retentionHours: 168, // 7 дней
    description: "Тестовые команды",
  },
  // ... другие топики
};

🧪 Как работает Request-Reply

1. test-sender-service → отправляет команду с correlation-id в test-commands
2. test-receiver-service → получает команду, обрабатывает через handleMessage()
3. test-receiver-service → отправляет ответ с тем же correlation-id в test-responses
4. test-sender-service → получает ответ и возвращает через Promise

Автоматически:

• Генерация correlation-id
• Добавление headers (correlation-id, reply-to, message-type, timestamp, а также дополнительные заголовки из additionalHeaders)
• Преобразование заголовков Kafka в Record<string, string> для передачи в handler
• Timeout обработка
• Отправка ответа с правильным correlation-id
• Обработка ошибок через RpcException с извлечением statusCode

📖 Низкоуровневый API (для reference)

Если нужен прямой доступ к KafkaCore:

import { KafkaCore } from '@packages/kafka-client';
import { LoggerService } from '@makebelieve21213-packages/logger';

const kafkaCore = new KafkaCore(
  {
    kafka: {
      clientId: 'test-service',
      brokers: ['kafka:9092'],
    },
    requestReply: {
      defaultTimeout: 30000,
      groupId: 'test-service-consumer',
    },
    retry: {
      maxRetries: 3,
      baseDelay: 1000,
      useExponentialBackoff: true,
    },
    dlq: {
      onMessage: async (payload) => {
        console.error('DLQ:', payload);
      },
      groupId: 'test-service-dlq-consumer',
    },
  },
  logger
);

await kafkaCore.connect();

// Fire-and-Forget
await kafkaCore.fireAndForget.send(KafkaTopic.TEST_COMMANDS, message);

// Request-Reply
kafkaCore.initRequestReply([KafkaTopic.TEST_RESPONSES]);
await kafkaCore.requestReply!.startListening();
const response = await kafkaCore.requestReply!.send(
  commandTopic, 
  responseTopic, 
  message,
  30000, // timeout (опционально)
  { 'user-id': '123' } // дополнительные заголовки (опционально)
);

🧪 Тестирование

Пакет имеет 100% покрытие тестами.

# Запустить тесты
pnpm test

# Запустить тесты с покрытием
pnpm test:coverage

# Watch режим
pnpm test:watch

🚨 Troubleshooting

Request timeout

Проблема: Request timeout after 30000ms

Решение:

1. Увеличить timeout: sendCommand(commandTopic, responseTopic, message, 60000)
2. Проверить, что test-service запущен и слушает топик
3. Проверить, что test-service отправляет ответ с правильным correlation-id
4. Проверить логи consumer'а

Сообщения не обрабатываются

Проблема: Сообщения отправляются, но не обрабатываются

Решение:

1. Проверить, что KafkaClientModule.forRootAsync() импортирован перед KafkaConsumerModule.forRoot()
2. Проверить, что consumer подписан на правильный топик
3. Проверить, что топик создан в Kafka
4. Проверить логи consumer'а
5. Проверить, что groupId уникален

Request-Reply не инициализирован

Проблема: Kafka Request-Reply not initialized

Решение:

1. Убедиться, что responseTopics указаны в KafkaClientModule.forRootAsync()
2. Проверить, что KafkaClientModule.forRootAsync() вызван перед KafkaProducerModule.forRootAsync()

Handler не получает зависимости

Проблема: Error: Nest can't resolve dependencies

Решение:

1. Добавить необходимые модули в imports опции KafkaConsumerModule.forRoot()
2. Убедиться, что handler'ы являются @Injectable() сервисами

Проблемы с подключением в Docker

Проблема: Не удается подключиться к Kafka из Docker контейнера

Решение:

1. Убедиться, что KAFKA_BROKERS указывает на правильный адрес (например, kafka:9092 внутри Docker сети)
2. Увеличить connectionTimeout и requestTimeout в опциях модуля
3. Проверить настройки retry для подключения
4. Убедиться, что Kafka доступен из контейнера

📊 Мониторинг

Проверка статуса подключения

const isConnected = kafkaProducer.isConnected();
console.log(`Kafka connected: ${isConnected}`);

Логирование

Все операции логируются через LoggerService из @makebelieve21213-packages/logger:

• Отправка команд
• Получение ответов
• Ошибки обработки
• Timeout события

🔧 Конфигурация Retry

{
  maxRetries: 3,                // Максимум попыток
  baseDelay: 1000,              // Базовая задержка (мс)
  useExponentialBackoff: true   // Exponential backoff
}

Exponential backoff: delay = baseDelay * 2^(retryCount - 1)

Пример:

• 1-я попытка: 1000ms
• 2-я попытка: 2000ms
• 3-я попытка: 4000ms

🔧 Конфигурация DLQ

{
  onMessage: async (payload) => {
    console.error('DLQ Message:', {
      originalTopic: payload.originalTopic,
      error: payload.error,
      failedAt: new Date(payload.failedAt),
    });
  },
  groupId: 'service-dlq-consumer'
}

📄 Лицензия

MIT

👥 Автор

Skryabin Aleksey