Revision history [back]

To turn off a consumer in the NestJS Kafka client, you can use the close() method provided by the Consumer object.

import { Controller } from '@nestjs/common';
import { MessagePattern } from '@nestjs/microservices';
import { Kafka } from 'kafkajs';

@Controller()
export class MyController {
  private kafka: Kafka;
  private consumer: Consumer;

  constructor() {
    this.kafka = new Kafka({
      clientId: 'my-app',
      brokers: ['localhost:9092'],
    });

    this.consumer = this.kafka.consumer({
      groupId: 'my-group',
    });

    this.consumer.connect();
  }

  @MessagePattern('my-topic')
  public async processMessage(payload: any) {
    console.log(`Received message: ${JSON.stringify(payload)}`);
    // process the message here
  }

  public async closeConsumer() {
    await this.consumer.disconnect();
  }
}

Once you're ready to turn off the consumer, simply call the closeConsumer() method. This will disconnect the consumer from the Kafka cluster and stop consuming messages.