feat(microservices): handle kafka consumer crashes

packages/microservices/client/client-kafka.ts

@@ -1,6 +1,7 @@
 import { Logger } from '@nestjs/common/services/logger.service';
 import { loadPackage } from '@nestjs/common/utils/load-package.util';
 import { isUndefined } from '@nestjs/common/utils/shared.utils';
+
 import {
   KAFKA_DEFAULT_BROKER,
   KAFKA_DEFAULT_CLIENT,
@@ -15,8 +16,10 @@ import {
   ConsumerConfig,
   ConsumerGroupJoinEvent,
   EachMessagePayload,
+  InstrumentationEvent,
   Kafka,
   KafkaConfig,
+  KafkaJSError,
   KafkaMessage,
   Producer,
   TopicPartitionOffsetAndMetadata,
@@ -142,6 +145,10 @@ export class ClientKafka extends ClientProxy {
             this.consumer.events.GROUP_JOIN,
             this.setConsumerAssignments.bind(this),
           );
+          this.consumer.on(
+            this.consumer.events.CRASH,
+            this.handleConsumerCrash,
+          );
           await this.consumer.connect();
           await this.bindTopics();
         }
@@ -330,4 +337,18 @@ export class ClientKafka extends ClientProxy {
       throw new Error('No consumer initialized');
     }
   }
+
+  private handleConsumerCrash(
+    event: InstrumentationEvent<{
+      error: KafkaJSError;
+      groupId: string;
+      restart: boolean;
+    }>,
+  ) {
+    const { error, groupId, restart } = event.payload;
+
+    if (!restart) {
+      throw new Error('Consumer crashed');
+    }
+  }
 }

packages/microservices/server/server-kafka.ts

@@ -1,6 +1,7 @@
 import { Logger } from '@nestjs/common/services/logger.service';
 import { isNil } from '@nestjs/common/utils/shared.utils';
 import { isObservable, lastValueFrom, Observable, ReplaySubject } from 'rxjs';
+
 import {
   KAFKA_DEFAULT_BROKER,
   KAFKA_DEFAULT_CLIENT,
@@ -17,8 +18,10 @@ import {
   Consumer,
   ConsumerConfig,
   EachMessagePayload,
+  InstrumentationEvent,
   Kafka,
   KafkaConfig,
+  KafkaJSError,
   KafkaMessage,
   Message,
   Producer,
@@ -103,6 +106,8 @@ export class ServerKafka extends Server implements CustomTransportStrategy {
     this.consumer = this.client.consumer(consumerOptions);
     this.producer = this.client.producer(this.options.producer);
 
+    this.consumer.on(this.consumer.events.CRASH, this.handleConsumerCrash);
+
     await this.consumer.connect();
     await this.producer.connect();
     await this.bindEvents(this.consumer);
@@ -317,4 +322,18 @@ export class ServerKafka extends Server implements CustomTransportStrategy {
   protected initializeDeserializer(options: KafkaOptions['options']) {
     this.deserializer = options?.deserializer ?? new KafkaRequestDeserializer();
   }
+
+  private handleConsumerCrash(
+    event: InstrumentationEvent<{
+      error: KafkaJSError;
+      groupId: string;
+      restart: boolean;
+    }>,
+  ) {
+    const { error, groupId, restart } = event.payload;
+
+    if (!restart) {
+      throw new Error('Consumer crashed');
+    }
+  }
 }

packages/microservices/test/client/client-kafka.spec.ts

@@ -1,5 +1,6 @@
 import { expect } from 'chai';
 import * as sinon from 'sinon';
+
 import { ClientKafka } from '../../client/client-kafka';
 import { NO_MESSAGE_HANDLER } from '../../constants';
 import { KafkaHeaders } from '../../enums';
@@ -176,6 +177,7 @@ describe('ClientKafka', () => {
         run,
         events: {
           GROUP_JOIN: 'consumer.group_join',
+          CRASH: 'consumer.crash',
         },
         on,
       };
@@ -288,7 +290,7 @@ describe('ClientKafka', () => {
 
         expect(consumerStub.calledOnce).to.be.true;
 
-        expect(on.calledOnce).to.be.true;
+        expect(on.calledTwice).to.be.true;
         expect(client['consumerAssignments']).to.be.empty;
 
         expect(connect.calledTwice).to.be.true;

packages/microservices/test/server/server-kafka.spec.ts

@@ -1,6 +1,7 @@
 import { Logger } from '@nestjs/common';
 import { AssertionError, expect } from 'chai';
 import * as sinon from 'sinon';
+
 import { NO_MESSAGE_HANDLER } from '../../constants';
 import { KafkaHeaders } from '../../enums';
 import {
@@ -90,6 +91,7 @@ describe('ServerKafka', () => {
   let subscribe: sinon.SinonSpy;
   let run: sinon.SinonSpy;
   let send: sinon.SinonSpy;
+  let on: sinon.SinonSpy;
   let consumerStub: sinon.SinonStub;
   let producerStub: sinon.SinonStub;
   let client;
@@ -101,12 +103,17 @@ describe('ServerKafka', () => {
     subscribe = sinon.spy();
     run = sinon.spy();
     send = sinon.spy();
+    on = sinon.spy();
 
     consumerStub = sinon.stub(server as any, 'consumer').callsFake(() => {
       return {
         connect,
         subscribe,
         run,
+        events: {
+          CRASH: 'consumer.crash',
+        },
+        on,
       };
     });
     producerStub = sinon.stub(server as any, 'producer').callsFake(() => {