nestjs/nest · error · InvalidKafkaClientTopicException

The client consumer did not subscribe to the corresponding r

Error message

The client consumer did not subscribe to the corresponding reply topic (${topic}).

What it means

Thrown as InvalidKafkaClientTopicException from getReplyTopicPartition() when a request-response (send()) call is dispatched but the reply topic (pattern + '.reply') is not present in this consumer's consumerAssignments map. consumerAssignments is populated from the KafkaJS group join event (setConsumerAssignments), so the error means the consumer group has not been assigned any partition for the reply topic that the producer headers asked the responder to use. Without an assigned partition the client cannot pin REPLY_PARTITION and cannot route the response back to this consumer.

Source

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

    const pattern = this.normalizePattern(packet.pattern);
    const outgoingEvent = await this.serializer.serialize(packet.data, {
      pattern,
    });
    const message = Object.assign(
      {
        topic: pattern,
        messages: [outgoingEvent],
      },
      this.options.send || {},
    );

    return this._producer!.send(message);
  }

  protected getReplyTopicPartition(topic: string): string {
    const minimumPartition = this.consumerAssignments[topic];
    if (isUndefined(minimumPartition)) {
      throw new InvalidKafkaClientTopicException(topic);
    }

    // Get the minimum partition
    return minimumPartition.toString();
  }

  protected publish(
    partialPacket: ReadPacket,
    callback: (packet: WritePacket) => any,
  ): () => void {
    const packet = this.assignPacketId(partialPacket);
    this.routingMap.set(packet.id, callback);

    const cleanup = () => this.routingMap.delete(packet.id);
    const errorCallback = (err: unknown) => {
      cleanup();
      callback({ err });
    };

View on GitHub (pinned to 6ec0e2783d)

Solutions

1. Ensure the client subscribes to the reply topic for each pattern you send() against (subscribeToResponseOf(pattern) or rely on the built-in auto-subscription to pattern + '.reply').
2. Wait for the consumer group to finish a rebalance before issuing send() calls — observe client.status for KafkaStatus.CONNECTED and a short warm-up, or await client.connect().
3. If you use a custom replyTopic, make sure the consumer's subscription set and the REPLY_TOPIC header produced by getResponsePatternName agree exactly; do not mix a custom reply topic with the default '.reply' suffix logic.
4. For request-response with Kafka, confirm the broker assigned partitions to this consumer for the reply topic (check the group join event / memberAssignment).

Example fix

// before
const client = new ClientKafka({ options: { client: { ... }, consumer: { groupId: 'svc' } } });
client.send('get-user', { id: 1 }).subscribe(); // may throw before group join

// after
await client.connect();
client.subscribeToResponseOf('get-user'); // subscribes to 'get-user.reply'
await client.connect(); // ensure consumer assignment is ready
client.send('get-user', { id: 1 }).subscribe();

// Validate reply-topic subscription readiness before send()
async function safeSend(client: ClientKafka, pattern: string, data: any) {
  const replyTopic = `${pattern}.reply`;
  const assignments = client.getConsumerAssignments?.() ?? {};
  if (!(replyTopic in assignments)) {
    // subscribe + wait for group join
    client.subscribeToResponseOf(pattern);
    await new Promise<void>(res => {
      const sub = client.status.subscribe(s => {
        if (s === KafkaStatus.CONNECTED) { setTimeout(() => { sub.unsubscribe(); res(); }, 100); }
      });
    });
  }
  return client.send(pattern, data);
}

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

try {
  return client.send(pattern, data);
} catch (e) {
  if (/did not subscribe to the corresponding reply topic/.test(e?.message)) {
    client.subscribeToResponseOf(pattern);
    await client.connect();
    return client.send(pattern, data);
  }
  throw e;
}

When it happens

Trigger: Calling client.send(pattern, data) (request-response) on a ClientKafka whose consumer has not subscribed to the reply topic, or before the first ConsumerGroupJoin event has fired and populated consumerAssignments. Subscribing only to request topics but not configuring the reply-topic subscription. Using a replyTopic name that differs from getResponsePatternName (pattern + '.reply'). Race condition: calling send() immediately after create() before group rebalance completes.

Common situations: subscribeToResponseOf() was not called for the reply pattern during client setup, or the auto-subscription to '<pattern>.reply' was disabled/overridden. Partition assignment latency right after a broker restart or rebalance. Custom serializer/deserializer that changes the reply-topic naming convention so the consumer subscribes to a topic that does not match what getResponsePatternName produces.

Related errors

• Method is not supported for Kafka client
• No consumer initialized. Please, call the "connect" method f
• No producer initialized. Please, call the "connect" method f
• No consumer initialized
• Not initialized. Please call the "connect" method first.

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