Kafka Producer with NestJS

This guide outlines how to set up a Kafka producer within a NestJS application. It explains how to configure Kafka, integrate it into a NestJS application, and create a controller to send messages to a Kafka topic.

Prerequisites

• Node.js installed on your machine.
• Kafka server running (locally or remotely).
• NestJS CLI installed. If not, install it via npm i -g @nestjs/cli.

Setup

Step 1: Create NestJS Project

If you haven't already, create a new NestJS project by running:

nest new project-name

Navigate to your project directory:

cd project-name

Step 2: Install Dependencies

Install the necessary dependencies for working with Kafka in NestJS:

npm install @nestjs/microservices kafkajs tokenx-kafka

Step 3: Kafka Configuration

Create a new file named kafka.config.ts in the root of your project. This file will contain the Kafka client configuration.

import { KafkaOptions, Transport } from '@nestjs/microservices';

export const kafkaConfig: KafkaOptions = {
  transport: Transport.KAFKA,
  options: {
    client: {
      brokers: ['localhost:9092'], // Adjust the broker address accordingly
      clientId: 'your-client-id',
    },
  },
};

Step 4: Main Application Setup

Open the main.ts file and modify it to connect the microservice (Kafka) with the main application.

import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module';
import { kafkaConfig } from './kafka.config';

async function bootstrap() {
  const app = await NestFactory.create(AppModule);

  app.connectMicroservice(kafkaConfig);

  await app.startAllMicroservicesAsync();
  await app.listen(3000);

  console.log(`Application is running on: 3000`);
}

bootstrap();

Step 5: Application Module Configuration

Open or create the app.module.ts file and set up the Kafka module and service.

import { Module } from '@nestjs/common';
import { AppController } from './app.controller';
import { ClientKafka } from '@nestjs/microservices';
import { kafkaConfig } from './kafka.config';
import { KafkaModule, KafkaService } from 'tokenx-kafka';

@Module({
  imports: [KafkaModule.forRootAsync(kafkaConfig)],
  controllers: [AppController],
  providers: [KafkaService, ClientKafka],
})
export class AppModule {}

Step 6: Create a Controller to Send Messages

Create a new file named app.controller.ts for your controller. This controller will handle HTTP POST requests to send messages to a Kafka topic.

import { Controller, Post, Body } from '@nestjs/common';
import { KafkaService } from 'tokenx-kafka';

@Controller()
export class AppController {
  constructor(private readonly kafkaService: KafkaService) {}

  @Post('/send')
  async sendMessage(@Body() message: any) {
    await this.kafkaService.sendMessage('my-first-topic', message);
    return { message: 'Message sent successfully', data: message };
  }
}

Replace 'my-first-topic' with the name of your Kafka topic.

Running the Application

Ensure your Kafka broker is running, then start your NestJS application by running:

npm run start

You can now send HTTP POST requests to http://localhost:3000/send with a JSON body, and the message will be forwarded to your Kafka topic.

Conclusion

You have successfully set up a Kafka producer within a NestJS application. This setup allows your application to send messages to a specified Kafka topic, integrating seamlessly with the broader Kafka ecosystem for real-time data processing and streaming.

Kafka Consumer with NestJS

This guide describes how to configure a Kafka consumer within a NestJS application. It covers the steps for setting up the Kafka connection, integrating it into your NestJS application, and creating a controller to handle messages from a Kafka topic.

Prerequisites

• Node.js installed on your system.
• A Kafka server running (either locally or remotely).
• The NestJS CLI installed. If not, you can install it via npm i -g @nestjs/cli.

Setup

Step 1: Create NestJS Project

If you haven't created a NestJS project yet, start by creating one:

nest new project-name

After creation, navigate to your project directory:

cd project-name

Step 2: Install Dependencies

Install the necessary dependencies for Kafka integration in NestJS:

npm install @nestjs/microservices kafkajs tokenx-kafka

Step 3: Kafka Configuration

Create a kafka.config.ts file in the root of your project for the Kafka client configuration, including the consumer group details.

import { KafkaOptions, Transport } from '@nestjs/microservices';

export const kafkaConfig: KafkaOptions = {
  transport: Transport.KAFKA,
  options: {
    client: {
      brokers: ['localhost:9092'], // Adjust according to your Kafka broker address
      clientId: 'your-client-id',
    },
    consumer: {
      groupId: 'my-consumer-group', // Define your consumer group
    },
  },
};

Step 4: Main Application Setup

Modify the main.ts file to connect the microservice (Kafka) with your application, and listen on a different port if needed.

import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module';
import { kafkaConfig } from './kafka.config';

async function bootstrap() {
  const app = await NestFactory.create(AppModule);

  app.connectMicroservice(kafkaConfig);

  await app.startAllMicroservicesAsync();
  await app.listen(3001); // Listen on a different port if your producer is on 3000

  console.log(`Application is running on: ${await app.getUrl()}`);
}

bootstrap();

Step 5: Application Module Configuration

In the app.module.ts file, set up the Kafka module, service, and the controller that will handle Kafka messages.

import { Module } from '@nestjs/common';
import { ClientKafka } from '@nestjs/microservices';
import { kafkaConfig } from './kafka.config';
import { KafkaController } from './kafka.controller';
import { KafkaModule, KafkaService } from 'tokenx-kafka';

@Module({
  imports: [KafkaModule.forRootAsync(kafkaConfig)],
  controllers: [KafkaController],
  providers: [KafkaService, ClientKafka],
})
export class AppModule {}

Step 6: Create a Controller to Handle Kafka Messages

Create a kafka.controller.ts file for your controller, which will handle messages from a specified Kafka topic.

import { Controller } from '@nestjs/common';
import { EventPattern, Payload } from '@nestjs/microservices';

@Controller()
export class KafkaController {
  @EventPattern('my-first-topic') // Listen to this topic
  async handleKafkaMessage(@Payload() message: any) {
    console.log(`Received message: ${JSON.stringify(message)}`);
    // Implement your processing logic here
  }
}

Ensure the topic name ('my-first-topic') matches the topic you wish to consume messages from.

Running the Application

Make sure your Kafka broker is running. Start your NestJS application:

npm run start

Your application is now set up as a Kafka consumer and will log messages received on the specified topic.

Conclusion

You have successfully set up a Kafka consumer within a NestJS application. This setup allows your application to consume messages from a specified Kafka topic, enabling real-time data processing and integration with the Kafka ecosystem.

Integrating Kafka Producer and Consumer with NestJS

This comprehensive guide demonstrates how to integrate both Kafka producer and consumer functionalities within a single NestJS application. It walks you through setting up Kafka, configuring your NestJS application, and implementing controllers for sending and receiving messages through Kafka topics.

Prerequisites

• Node.js installed on your machine.
• A Kafka server running locally or remotely.
• NestJS CLI installed (npm i -g @nestjs/cli).

Project Setup

1. Create NestJS Project

Start by creating a new NestJS project if you haven't already:

nest new kafka-nestjs-project

Navigate to the project directory:

cd kafka-nestjs-project

2. Install Dependencies

Install the necessary packages for Kafka integration:

npm install @nestjs/microservices kafkajs tokenx-kafka

Kafka Configuration

Create a kafka.config.ts file in your project's root directory. This file will contain configuration for both the Kafka producer and consumer.

import { KafkaOptions, Transport } from '@nestjs/microservices';

export const kafkaConfig: KafkaOptions = {
  transport: Transport.KAFKA,
  options: {
    client: {
      brokers: ['localhost:9092'], // Adjust based on your Kafka broker addresses
      clientId: 'your-client-id',
    },
    consumer: {
      groupId: 'my-consumer-group', // Define a consumer group for your application
    },
  },
};

Application Setup

Main Application File (main.ts)

Configure your application to connect to Kafka in the main.ts file. Here, you will start both the microservice and the HTTP server.

import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module';
import { kafkaConfig } from './kafka.config';

async function bootstrap() {
  const app = await NestFactory.create(AppModule);

  app.connectMicroservice(kafkaConfig);

  await app.startAllMicroservicesAsync();
  await app.listen(3000);
  console.log(`Application is running on: ${await app.getUrl()}`);
}

bootstrap();

Application Module (app.module.ts)

In app.module.ts, import your Kafka configuration and controllers:

import { Module } from '@nestjs/common';
import { ClientKafka } from '@nestjs/microservices';
import { kafkaConfig } from './kafka.config';
import { AppController } from './app.controller';
import { KafkaController } from './kafka.controller';
import { KafkaModule, KafkaService } from 'tokenx-kafka';

@Module({
  imports: [KafkaModule.forRootAsync(kafkaConfig)],
  controllers: [AppController, KafkaController],
  providers: [KafkaService, ClientKafka],
})
export class AppModule {}

Implementing Kafka Producer and Consumer

Producer Controller (app.controller.ts)

This controller will send messages to a Kafka topic upon receiving an HTTP POST request.

import { Controller, Post, Body } from '@nestjs/common';
import { KafkaService } from 'tokenx-kafka';

@Controller()
export class AppController {
  constructor(private readonly kafkaService: KafkaService) {}

  @Post('/send')
  async sendMessage(@Body() message: any) {
    await this.kafkaService.sendMessage('my-first-topic', message);
    return { message: 'Message sent successfully', data: message };
  }
}

Consumer Controller (kafka.controller.ts)

Create a controller to handle messages from your Kafka topic.

import { Controller } from '@nestjs/common';
import { EventPattern, Payload } from '@nestjs/microservices';

@Controller()
export class KafkaController {
  @EventPattern('my-first-topic')
  async handleKafkaMessage(@Payload() message: any) {
    console.log(`Received message: ${JSON.stringify(message)}`);
    // Implement your message handling logic here
  }
}

Running the Application

Make sure your Kafka broker is up and running. Start your NestJS application:

npm run start

Your application can now send messages to and consume messages from Kafka topics.

Conclusion

You've successfully integrated Kafka producer and consumer functionalities in a single NestJS application. This setup enables your application to participate in real-time data processing and messaging workflows, leveraging Kafka's robust messaging system.