nestjs/nest · error · Error

No consumer initialized

Error message

No consumer initialized

What it means

`ClientKafka.commitOffsets()` throws a bare `Error('No consumer initialized')` when `_consumer` is null — before `connect()` completes or in `producerOnlyMode`. Manual offset commits require a live consumer; this guard prevents a null dereference deeper in kafkajs.

Source

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

    }
    const source = defer(async () => this.connect()).pipe(
      mergeMap(() => this.dispatchBatchEvent({ pattern, data })),
    );
    const connectableSource = connectable(source, {
      connector: () => new Subject(),
      resetOnDisconnect: false,
    });
    connectableSource.connect();
    return connectableSource;
  }

  public commitOffsets(
    topicPartitions: TopicPartitionOffsetAndMetadata[],
  ): Promise<void> {
    if (this._consumer) {
      return this._consumer.commitOffsets(topicPartitions);
    } 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');
  }

View on GitHub (pinned to 6ec0e2783d)

Solutions

1. Await `connect()` before calling `commitOffsets`.
2. Disable `producerOnlyMode` if you need a consumer for manual commits.
3. Gate manual commits behind a `getConsumerAssignments()` readiness check.

Example fix

// before
await this.client.commitOffsets([{ topic: 'users', partition: 0, offset: '42' }]); // throws
// after
await this.client.connect();
await this.client.commitOffsets([{ topic: 'users', partition: 0, offset: '42' }]);

// Only commit offsets once a consumer exists and has joined
await this.client.connect();
const assignments = this.client.getConsumerAssignments();
if (Object.keys(assignments).length > 0) {
  await this.client.commitOffsets([{ topic: 'users', partition: 0, offset: '42' }]);
}

function hasConsumer(c: ClientKafka): boolean { return Reflect.get(c, '_consumer') != null; }

try {
  await this.client.commitOffsets(offsets);
} catch (e) {
  if (e instanceof Error && /No consumer initialized/.test(e.message)) { /* defer retry */ }
  throw e;
}

When it happens

Trigger: Calling `clientKafka.commitOffsets([{ topic, partition, offset }])` before `connect()` resolves, or while running with `producerOnlyMode: true`.

Common situations: At-least-once consumer trying to commit offsets manually before the consumer has joined the group; producer-only client reused for offset management.

Related errors

• No consumer initialized. Please, call the "connect" method f
• No producer initialized. Please, call the "connect" method f
• Not initialized. Please call the "connect" method first.
• 3
• The "connect()" method is not supported in gRPC mode.

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