"Custom exception handling" example from NestJS Kafka docs doesn't work

[kacperbudny-pc]

Is there an existing issue for this?

• I have searched the existing issues

Current behavior

I tried implementing custom exception handling, but the example given in the docs doesn't work:

https://docs.nestjs.com/microservices/kafka#custom-exception-handling

The problem here is, when this exception filter runs, it fails with the following error:

ERROR [Consumer] Crash: KafkaJSNonRetriableError: Cannot read properties of undefined (reading 'isHeadersSent') 

I tried to look into the source code to find out why that happens, and it looks like BaseExceptionFilter assumes that we are in HTTP context, which we are not when working with Kafka. So I tried to re-throw the error instead, but this time it looks like the headers are always empty, so the retry count is always 0.

Minimum reproduction code

https://github.com/kacperbudny-pc/kafka-retries-example

Steps to reproduce

No response

Expected behavior

I can use the example from the docs to control the amount of retries

Package

• I don't know. Or some 3rd-party package
• @nestjs/common
• @nestjs/core
• @nestjs/microservices
• @nestjs/platform-express
• @nestjs/platform-fastify
• @nestjs/platform-socket.io
• @nestjs/platform-ws
• @nestjs/testing
• @nestjs/websockets
• Other (see below)

Other package

No response

NestJS version

11.0.1

Packages versions

    "@nestjs/common": "^11.0.1",
    "@nestjs/core": "^11.0.1",
    "@nestjs/microservices": "^11.0.13",
    "@nestjs/platform-express": "^11.0.1",
    "kafkajs": "^2.2.4",
    "reflect-metadata": "^0.2.2",
    "rxjs": "^7.8.1"

Node.js version

22.12.0

In which operating systems have you tested?

• macOS
• Windows
• Linux

Other

No response

[kamilmysliwiec]

cc @gkdis6 ref nestjs/docs.nestjs.com#3223

[gkdis6]

I’ve modified the kafka-exception-filter.ts file for debugging and have observed a slight discrepancy compared to the behavior mentioned in this issue.

According to the issue, when the Kafka consumer restarts due to a KafkaJSNumberOfRetriesExceeded error, the logged error messages include a retryCount property that tracks the number of failed attempts for a particular offset. For example, the issue notes an error log like:

ERROR [Consumer] Crash: KafkaJSNumberOfRetriesExceeded: invalid.relation.id {"timestamp":"2025-02-19T11:43:47.417Z","logger":"kafkajs","groupId":"my-server","retryCount":5,"stack":"KafkaJSNonRetriableError\n  Caused by: Error: ....}

In my tests, however, I noticed the following behavior:

• The consumer restart log reports that there were 6 processing attempts and the retryCount logged as 5:

[Nest] 23574  - 04/09/2025, 10:11:48 AM   ERROR [KafkaMaxRetryExceptionFilter] Error occurred during Kafka processing:
[Nest] 23574  - 04/09/2025, 10:11:48 AM   ERROR [KafkaMaxRetryExceptionFilter] Error: test
    at AppController.handleEvent (/Users/sangyongpark/code/kafka-retries-example/src/app.controller.ts:17:11)
    at /Users/sangyongpark/code/kafka-retries-example/node_modules/.pnpm/@nestjs+microservices@11.0.13_@nestjs+common@11.0.13_reflect-metadata@0.2.2_rxjs@7.8.2__27a92eeef769419be0f8695ddd81dd4d/node_modules/@nestjs/microservices/context/rpc-context-creator.js:44:33
    at process.processTicksAndRejections (node:internal/process/task_queues:105:5)
    at async /Users/sangyongpark/code/kafka-retries-example/node_modules/.pnpm/@nestjs+microservices@11.0.13_@nestjs+common@11.0.13_reflect-metadata@0.2.2_rxjs@7.8.2__27a92eeef769419be0f8695ddd81dd4d/node_modules/@nestjs/microservices/context/rpc-proxy.js:11:32
    at async ServerKafka.handleEvent (/Users/sangyongpark/code/kafka-retries-example/node_modules/.pnpm/@nestjs+microservices@11.0.13_@nestjs+common@11.0.13_reflect-metadata@0.2.2_rxjs@7.8.2__27a92eeef769419be0f8695ddd81dd4d/node_modules/@nestjs/microservices/server/server-kafka.js:226:32)
    at async Runner.processEachMessage (/Users/sangyongpark/code/kafka-retries-example/node_modules/.pnpm/kafkajs@2.2.4/node_modules/kafkajs/src/consumer/runner.js:231:9)
    at async onBatch (/Users/sangyongpark/code/kafka-retries-example/node_modules/.pnpm/kafkajs@2.2.4/node_modules/kafkajs/src/consumer/runner.js:447:9)
    at async Runner.handleBatch (/Users/sangyongpark/code/kafka-retries-example/node_modules/.pnpm/kafkajs@2.2.4/node_modules/kafkajs/src/consumer/runner.js:461:5)
    at async /Users/sangyongpark/code/kafka-retries-example/node_modules/.pnpm/kafkajs@2.2.4/node_modules/kafkajs/src/consumer/worker.js:29:9
[Nest] 23574  - 04/09/2025, 10:11:48 AM   ERROR [ServerKafka] ERROR [Runner] Error when calling eachMessage {"timestamp":"2025-04-09T01:11:48.113Z","logger":"kafkajs","topic":"test","partition":0,"offset":"35","stack":"Error: test\n    at AppController.handleEvent (/Users/sangyongpark/code/kafka-retries-example/src/app.controller.ts:17:11)\n    at /Users/sangyongpark/code/kafka-retries-example/node_modules/.pnpm/@nestjs+microservices@11.0.13_@nestjs+common@11.0.13_reflect-metadata@0.2.2_rxjs@7.8.2__27a92eeef769419be0f8695ddd81dd4d/node_modules/@nestjs/microservices/context/rpc-context-creator.js:44:33\n    at process.processTicksAndRejections (node:internal/process/task_queues:105:5)\n    at async /Users/sangyongpark/code/kafka-retries-example/node_modules/.pnpm/@nestjs+microservices@11.0.13_@nestjs+common@11.0.13_reflect-metadata@0.2.2_rxjs@7.8.2__27a92eeef769419be0f8695ddd81dd4d/node_modules/@nestjs/microservices/context/rpc-proxy.js:11:32\n    at async ServerKafka.handleEvent (/Users/sangyongpark/code/kafka-retries-example/node_modules/.pnpm/@nestjs+microservices@11.0.13_@nestjs+common@11.0.13_reflect-metadata@0.2.2_rxjs@7.8.2__27a92eeef769419be0f8695ddd81dd4d/node_modules/@nestjs/microservices/server/server-kafka.js:226:32)\n    at async Runner.processEachMessage (/Users/sangyongpark/code/kafka-retries-example/node_modules/.pnpm/kafkajs@2.2.4/node_modules/kafkajs/src/consumer/runner.js:231:9)\n    at async onBatch (/Users/sangyongpark/code/kafka-retries-example/node_modules/.pnpm/kafkajs@2.2.4/node_modules/kafkajs/src/consumer/runner.js:447:9)\n    at async Runner.handleBatch (/Users/sangyongpark/code/kafka-retries-example/node_modules/.pnpm/kafkajs@2.2.4/node_modules/kafkajs/src/consumer/runner.js:461:5)\n    at async /Users/sangyongpark/code/kafka-retries-example/node_modules/.pnpm/kafkajs@2.2.4/node_modules/kafkajs/src/consumer/worker.js:29:9","error":{}}
... (6 times)
ERROR [Consumer] Restarting the consumer in 8824ms {"timestamp":"2025-04-09T01:11:48.132Z","logger":"kafkajs","retryCount":5,"retryTime":8824,"groupId":"nestjs-group-server"}

• When testing a failed Kafka connection, the initial error logs for broker connection failures show the retryCount property accurately (incrementing from 0 to 5). Here’s a snippet of the logs observed during connection errors:

[Nest] 23574  - 04/09/2025, 10:18:16 AM   ERROR [ServerKafka] ERROR [BrokerPool] Failed to connect to seed broker, trying another broker from the list: Connection error: read ECONNRESET {"timestamp":"2025-04-09T01:18:16.943Z","logger":"kafkajs","retryCount":0,"retryTime":358}
[Nest] 23574  - 04/09/2025, 10:18:17 AM   ERROR [ServerKafka] ERROR [Connection] Connection error:  {"timestamp":"2025-04-09T01:18:17.304Z","logger":"kafkajs","broker":"localhost:9092","clientId":"nestjs-consumer-server","stack":"AggregateError [ECONNREFUSED]: \n    at internalConnectMultiple (node:net:1139:18)\n    at afterConnectMultiple (node:net:1712:7)"}
[Nest] 23574  - 04/09/2025, 10:18:17 AM   ERROR [ServerKafka] ERROR [BrokerPool] Failed to connect to seed broker, trying another broker from the list: Connection error:  {"timestamp":"2025-04-09T01:18:17.304Z","logger":"kafkajs","retryCount":1,"retryTime":624}
[Nest] 23574  - 04/09/2025, 10:18:17 AM   ERROR [ServerKafka] ERROR [Connection] Connection error:  {"timestamp":"2025-04-09T01:18:17.930Z","logger":"kafkajs","broker":"localhost:9092","clientId":"nestjs-consumer-server","stack":"AggregateError [ECONNREFUSED]: \n    at internalConnectMultiple (node:net:1139:18)\n    at afterConnectMultiple (node:net:1712:7)"}
[Nest] 23574  - 04/09/2025, 10:18:17 AM   ERROR [ServerKafka] ERROR [BrokerPool] Failed to connect to seed broker, trying another broker from the list: Connection error:  {"timestamp":"2025-04-09T01:18:17.930Z","logger":"kafkajs","retryCount":2,"retryTime":1342}
... (logs showing retryCount progressing to 5)

Despite these accurate logs for connection errors, the error logs during message processing (from the KafkaMaxRetryExceptionFilter) do not include the retryCount property, making it unclear how to verify the actual retry number. At the moment, it’s uncertain if the omission in the log is due to the way the exception filter handles the error object or if KafkaJS is not propagating the value as expected in that context.

I plan to dig deeper to determine whether this behavior is intentional or if an improvement is needed in logging to expose the retry count during message processing.

[gkdis6]

The previous filter is already merged, and I believe most people here are familiar with that implementation.

In contrast, this filter focuses on handling application-level message processing errors, rather than connection-level retry failures. To that end, I experimented with a circuit breaker–style exception filter to prevent repeatedly failing messages from overwhelming the system and allow recovery after a cooldown period.

Also, unlike the existing implementation which extends BaseExceptionFilter, this filter implements ExceptionFilter directly, allowing for more flexibility and decoupling from Nest’s default global exception handling behavior. This is particularly useful for message-driven systems like Kafka where retry logic and message acknowledgment need to be carefully controlled.

Here is the full code for reference:

import { Catch, ArgumentsHost, Logger, ExceptionFilter } from '@nestjs/common';
import { KafkaContext } from '@nestjs/microservices';

@Catch()
export class KafkaMaxRetryExceptionFilter implements ExceptionFilter {
  private readonly logger = new Logger(KafkaMaxRetryExceptionFilter.name);
  private failureCount = 0;
  private circuitOpen = false;
  private openTimestamp = 0;
  private readonly circuitResetTimeout = 30000;

  constructor(
    private readonly maxRetries: number,
    private readonly skipHandler?: (message: any) => Promise<void>,
  ) {}

  async catch(exception: unknown, host: ArgumentsHost) {
    const kafkaContext = host.switchToRpc().getContext<KafkaContext>();
    const message = kafkaContext.getMessage();

    if (this.circuitOpen) {
      if (Date.now() - this.openTimestamp >= this.circuitResetTimeout) {
        this.logger.log('Circuit breaker timeout elapsed. Resetting breaker.');
        this.circuitOpen = false;
        this.failureCount = 0;
      } else {
        this.logger.warn('Circuit breaker open. Skipping message processing.');
        await this.commitOffset(kafkaContext);
        return;
      }
    }

    this.failureCount++;
    this.logger.log('failureCount', { failureCount: this.failureCount });

    if (this.failureCount >= this.maxRetries) {
      this.logger.warn(`Circuit breaker triggered after ${this.failureCount} failures. Skipping message processing.`);
      if (this.skipHandler) {
        try {
          await this.skipHandler(message);
        } catch (err) {
          this.logger.error('Error in skipHandler:', err);
        }
      }
      try {
        await this.commitOffset(kafkaContext);
      } catch (commitError) {
        this.logger.error('Failed to commit offset:', commitError);
      }
      this.circuitOpen = true;
      this.openTimestamp = Date.now();
      return;
    }

    throw exception;
  }

  private getRetryCountFromContext(context: KafkaContext): number {
    const headers = context.getMessage().headers || {};
    this.logger.log('headers', { headers });
    const retryHeader = headers['retryCount'] || headers['retry-count'];
    return retryHeader ? Number(retryHeader) : 0;
  }

  private async commitOffset(context: KafkaContext): Promise<void> {
    const consumer = context.getConsumer && context.getConsumer();
    if (!consumer) {
      throw new Error('Consumer instance is not available from KafkaContext.');
    }
    const topic = context.getTopic && context.getTopic();
    const partition = context.getPartition && context.getPartition();
    const message = context.getMessage();
    const offset = message.offset;
    if (!topic || partition === undefined || offset === undefined) {
      throw new Error('Incomplete Kafka message context for committing offset.');
    }
    await consumer.commitOffsets([
      {
        topic,
        partition,
        offset: (Number(offset) + 1).toString(),
      },
    ]);
  }
}

This filter is designed to handle repeated processing failures gracefully, providing an alternative to header-based retries by skipping problematic messages and temporarily halting message processing when necessary.

Because the intent of this filter is clearly distinct from the previously merged one—that focuses on retrying based on Kafka headers—I believe it makes sense to maintain them as separate implementations rather than merging them into one.

Let me know if any clarification or adjustments are needed!

[kamilmysliwiec]

Do you think we should remove that section from the docs then, @gkdis6? also btw since your new suggested implementation manually counts the number of failures, registering a global filter for all handlers won't work (you'd need a dedicated filter instance per handler)

[gkdis6]

I believe it would still be valuable to include guidance, even if we present two separate examples for different types of filters. While allowing users to choose between registering a global filter and assigning a dedicated filter instance per handler would be ideal, there could also be legitimate use cases where finer-grained control is necessary—for instance, applying stricter or more relaxed rules on a per-handler basis.

[kamilmysliwiec]

But if I understand correctly, the current implementation proposed in the docs is not valid (according to the minimum reproduction repository)

[gkdis6]

The initial implementation I proposed in the PR properly handled retries only for connection errors, whereas the version I shared in the recent comment is only valid for message processing errors. Since neither of them fully covers both cases, I believe it makes sense to separate their usage. Also, I think the guide should serve as a minimal reference—showing that “this is one possible approach”—rather than trying to cover every use case. That’s why I believe presenting two distinct examples is even more important.

[kamilmysliwiec]

Would you like to create a PR to the docs that reflects what we discussed here? With 2 examples covering 2 different use-cases

[gkdis6]

Of course! I’ll double-check to see if there’s any way to cover both cases in a single approach, and also make sure we haven’t missed any edge cases. I’ll aim to put the PR together by April 13.

[gkdis6]

I apologize for any confusion caused by the previous approach.

Issue Description
In a NestJS microservices environment using Kafka, Exception Filters are designed to handle errors that occur within HTTP requests/responses or within message pattern handlers. However, Kafka connection errors (e.g., ECONNREFUSED) are thrown by KafkaJS during its asynchronous connection phase—before an HTTP context is established. As a result, these errors do not reach the Exception Filter.

In our test cases, the connection attempt is made at the code level and the error eventually reaches the filter. In a real service environment, however, the error occurs before establishing the connection and is never passed to the filter.

Proposed Solution
To resolve this issue, we propose using KafkaJS’s retry option to limit the number of retry attempts when connecting to the Kafka broker. See the example code below:

const microserviceOptions: MicroserviceOptions = {
  transport: Transport.KAFKA,
  options: {
    client: {
      brokers: ['localhost:9092'],
      retry: {
        // Initial retry delay in milliseconds
        initialRetryTime: 300,
        // Maximum number of retries
        retries: 4,
        // Multiplier for increasing wait time between retries (e.g., 2x)
        factor: 2,
      },
    },
  },
};

app.connectMicroservice<MicroserviceOptions>(microserviceOptions);

With this configuration, if the connection to the Kafka broker fails, it will retry up to 4 times, thereby reducing unnecessary connection attempts. This approach effectively handles early-stage errors that are not captured by the Exception Filter.

[kamilmysliwiec]

@gkdis6 How's this relevant to the issue that @kacperbudny-pc described?