nestjs/nest · error · Error

Method is not supported for Kafka client

Error message

Method is not supported for Kafka client

What it means

Thrown by ClientKafka.on() because the Kafka client intentionally does not expose the generic event-listener API that the other transport clients (TCP, Redis, RMQ, MQTT) provide. Kafka event listening is handled internally via registerConsumerEventListeners()/registerProducerEventListeners() and exposed through the status Observable and a typed Kafka-specific on() override surface. Calling on() directly on a ClientKafka instance is an unsupported operation by design, not a connection problem.

Source

Thrown at packages/microservices/client/client-kafka.ts:296

    } else {
      throw new Error('No consumer initialized');
    }
  }

  public unwrap<T>(): T {
    if (!this.client) {
      throw new Error(
        'Not initialized. Please call the "connect" method first.',
      );
    }
    return this.client as T;
  }

  public on<
    EventKey extends string | number | symbol = string | number | symbol,
    EventCallback = any,
  >(event: EventKey, callback: EventCallback) {
    throw new Error('Method is not supported for Kafka client');
  }

  protected registerConsumerEventListeners() {
    if (!this._consumer) {
      return;
    }
    this._consumer.on(this._consumer.events.CONNECT, () =>
      this._status$.next(KafkaStatus.CONNECTED),
    );
    this._consumer.on(this._consumer.events.DISCONNECT, () =>
      this._status$.next(KafkaStatus.DISCONNECTED),
    );
    this._consumer.on(this._consumer.events.REBALANCING, () =>
      this._status$.next(KafkaStatus.REBALANCING),
    );
    this._consumer.on(this._consumer.events.STOP, () =>
      this._status$.next(KafkaStatus.STOPPED),
    );

View on GitHub (pinned to 6ec0e2783d)

Solutions

  1. Subscribe to the client.status Observable instead: client.status.subscribe(s => ...) emits KafkaStatus.CONNECTED/DISCONNECTED/REBALANCING/STOPPED/CRASHED.
  2. If you need raw consumer/producer events, call client.unwrap() after connect to get the underlying KafkaJS consumer/producer and attach listeners there.
  3. Remove any client.on(...) call against a ClientKafka instance; this path is unsupported and cannot be enabled via configuration.

Example fix

// before
const client = ClientProxyFactory.create({ transport: Transport.KAFKA, options: {...} });
client.on('connect', () => console.log('connected'));

// after
client.status.subscribe(status => {
  if (status === KafkaStatus.CONNECTED) console.log('connected');
});
Defensive patterns

Strategy: type-guard

Validate before calling

import { ClientKafka } from '@nestjs/microservices';

function supportsOn(client: any): boolean {
  // ClientKafka.on() always throws 'Method is not supported for Kafka client'
  return !(client instanceof ClientKafka);
}

if (supportsOn(client)) client.on('connect', fn);
else client.status.subscribe(s => { /* ... */ });

Type guard

import { ClientKafka } from '@nestjs/microservices';
const isKafkaClient = (c: unknown): c is ClientKafka => c instanceof ClientKafka;

Try / catch

if (client instanceof ClientKafka) {
  client.status.subscribe(s => onStatus(s));
} else {
  client.on('connect', onConnect);
}

Prevention

When it happens

Trigger: Calling client.on('connect', fn), client.on('disconnect', fn), or any client.on(event, callback) on an instance of ClientKafka (the object returned by ClientProxyFactory.create({ transport: Transport.KAFKA }) or ClientsModule). Any code ported from a Redis/TCP client that uses on() to listen for broker events will hit this on the Kafka transport.

Common situations: Migrating a microservice from Redis/RMQ/TCP transport to Kafka and copying the existing client.on(...) status wiring verbatim. Generic connection-monitoring code that loops over transports and calls on() uniformly. Using a shared base class that calls on() assuming all clients support it.

Related errors

AI-assisted analysis of nestjs/nest@6ec0e2783d (2026-08-03). Data as JSON: /data/errors/283fc37085203da8.json. Report an issue: GitHub.