Kafka integration helpers

Nest-friendly wrappers around kafkajs to configure Kafka clients once and inject typed consumers, producers, admins, and loggers throughout your modules.

Installation

Install both required dependencies:

npm install @nestjs/common kafkajs

Or with other package managers:

pnpm add @nestjs/common kafkajs
yarn add @nestjs/common kafkajs
bun add @nestjs/common kafkajs

Tokens and decorators

• Tokens are scoped by Kafka name and client name: getKafkaToken(name), getConsumerToken(consumer, kafka), getProducerToken(producer, kafka), getAdminToken(admin, kafka), getLoggerToken(kafka).
• Use decorators to inject: @InjectKafka(name?), @InjectConsumer(consumer?, kafka?), @InjectProducer(producer?, kafka?), @InjectAdmin(admin?, kafka?), @InjectLogger(kafka?).
• Default names: Kafka default_kafka, consumer default_consumer, producer default_producer, admin default_admin.

Registering Kafka

KafkaModule.forRoot({
  kafka: "primary", // optional
  config: { clientId: "chat-service", brokers: ["localhost:9092"] },
});

KafkaModule.forRootAsync({
  kafka: "analytics",
  imports: [ConfigModule],
  useFactory: (config: ConfigService) => ({
    kafka: "analytics",
    config: { clientId: "analytics", brokers: config.get("KAFKA_BROKERS") },
  }),
  inject: [ConfigService],
});

Consumers (sync register)

KafkaModule.registerConsumers(
  [
    {
      name: "chat-consumer",
      config: { groupId: "chat-group" },
      autoConnect: true,
    },
  ],
  "primary",
);

Consumer (async)

KafkaModule.registerConsumerAsync(
  {
    name: "chat-consumer",
    kafka: "primary",
    useFactory: async () => ({
      name: "chat-consumer",
      config: { groupId: "chat-group" },
    }),
  },
);

Producers / Admins

APIs mirror consumers:

• registerProducers, registerProducerAsync
• registerAdmins, registerAdminAsync

Logger

registerLogger(kafkaName?) exposes one logger per Kafka instance (kafka.logger() under the hood). Inject with @InjectLogger(kafkaName?).

Lifecycle

• Consumers/producers/admins are disconnected automatically on Nest shutdown.