@application-kit/kafka

Kafka-транспорт для NestJS: бутстрап консьюмер-процесса с heartbeat-файлом, труба потребления (ручной commit offset'а, retry и DLQ), имена топиков и контракт версии события, общий для производителя (outbox + Debezium EventRouter) и потребителя.

Установка

yarn add @application-kit/kafka

Peer-зависимости: @nestjs/common, @nestjs/core, @nestjs/microservices, kafkajs, rxjs, @libs-for-dev/nestjs-config-module, class-transformer, class-validator, nestjs-pino. @nestjs/microservices и kafkajs нужны только корневому входу и объявлены опциональными: процессу, который берёт лишь @application-kit/kafka/event-contract, ставить их не нужно.

Пакет грузится одним экземпляром на процесс: DI-токены — Symbol(), и ESM- и CJS-копии разом дали бы два набора токенов.

Окружение

| Переменная | Назначение | |---|---| | KAFKA_BROKERS | список брокеров через запятую | | KAFKA_CLIENT_ID | префикс clientId (к нему добавляется uuid процесса) | | KAFKA_GROUP_ID | группа консьюмера — нужна только консьюмер-процессу | | KAFKA_LOG_LEVEL | уровень kafkajs: NOTHING, ERROR, WARN, INFO (дефолт), DEBUG | | KAFKA_SASL_USERNAME, KAFKA_SASL_PASSWORD | SCRAM-SHA-512; не заданы — без аутентификации | | KAFKA_SSL_CA_FILE | путь к CA брокера; не задан — без TLS | | HEARTBEAT_FILE | файл пульса для exec-liveness; не задан — пульс не пишется |

Переменные валидируются при старте: с неверным окружением процесс не поднимается. Импорт пакета окружения не требует — KAFKA_* читаются при вызове KafkaModule.forRootAsync и kafkaConsumerServer, поэтому производитель событий берёт из пакета versionedQueueName без них.

Консьюмер-процесс

import { KafkaLogger, KafkaModule, kafkaConsumerServer } from '@application-kit/kafka'
import { Module } from '@nestjs/common'
import { Logger } from 'nestjs-pino'

@Module({
  imports: [
    // logger — класс, а не экземпляр: экземпляр, созданный на импорте файла, забрал бы себе
    // корневой логгер nestjs-pino с дефолтными опциями.
    KafkaModule.forRootAsync({ logger: KafkaLogger }),
    // …модули с консьюмерами
  ],
})
class ConsumerModule {}

void kafkaConsumerServer(ConsumerModule, Logger)

kafkaConsumerServer поднимает Nest-микросервис и обновляет HEARTBEAT_FILE на событиях kafkajs HEARTBEAT и GROUP_JOIN: зависшая группа перестаёт слать пульс, файл протухает, и оркестратор перезапускает процесс по mtime.

forRootAsync принимает errorHandlerOptions — maxAttempts (5), baseDelayMs (500) и topicSuffix (retry, dlq). KafkaLogger — реализация KafkaLoggerInterface на nestjs-pino; свой логгер — любой класс с error, info, warn.

Консьюмер

import {
  ConsumerAbstract,
  eventIdOf,
  eventTopics,
  KafkaAutoCommit,
  type KafkaMessageContextInterface,
} from '@application-kit/kafka'

@Controller()
export class UserConsumer extends ConsumerAbstract {
  @EventPattern(eventTopics(UserWasCreatedEvent.queueName))
  @KafkaAutoCommit()
  public async created(@Payload() payload: UserDto, @Ctx() context: KafkaMessageContextInterface) {
    await this.handleAll(payload, [this.welcomeLetter, this.audit])
  }
}

• eventTopics(queue, version?) — пара топиков очереди: events.<queue> и его .retry.
• @KafkaAutoCommit() коммитит offset после успешной обработки; упавшее сообщение уходит в retry-топик с растущей задержкой, после maxAttempts — в DLQ.
• ConsumerAbstract.handleAll запускает обработчики (HandlerInterface<TPayload>) параллельно; отказ любого — ConsumersError, сообщение переигрывается целиком.
• eventIdOf(context) — идентификатор события из заголовка id (его ставит Debezium EventRouter).

Версия контракта события

Несовместимая смена контракта уводит события в свою очередь <queue>.v<N>; первая версия живёт в топике без сегмента. Производитель собирает имя очереди versionedQueueName(queue, version), потребитель подписывается eventTopics(queue, version), а труба сверяет версию топика (eventVersionOfTopic) с заголовком x-event-version (eventVersionOfHeaders) до вызова обработчика: сообщение чужой версии уходит в DLQ, не доходя до кода потребителя.

Правила версии живут и отдельным входом @application-kit/kafka/event-contract — без kafkajs, @nestjs/microservices и побочных эффектов. Его берёт производитель события (web-процесс, outbox), которому транспорт не нужен; корень пакета экспортирует то же самое.