feat(microservices): add kafka retriable exception, rethrow if needed

Author: kamilmysliwiec

packages/microservices/context/kafka-rpc-proxy.ts

@@ -1,5 +1,5 @@
 import { ExecutionContextHost } from '@nestjs/core/helpers/execution-context-host';
-import { Observable, isObservable } from 'rxjs';
+import { isObservable, Observable } from 'rxjs';
 import { catchError } from 'rxjs/operators';
 import { RpcExceptionsHandler } from '../exceptions/rpc-exceptions-handler';
 

packages/microservices/context/rpc-proxy.ts

@@ -35,7 +35,7 @@ export class KafkaContext extends BaseRpcContext<KafkaContextArgs> {
   }
 
   /**
-   * Commit the offset of this message.
+   * Returns the Kafka consumer reference.
    */
   getConsumer() {
     return this.args[3];

packages/microservices/ctx-host/kafka.context.ts

@@ -1,2 +1,3 @@
 export * from './base-rpc-exception-filter';
+export * from './kafka-retriable-exception';
 export * from './rpc-exception';

packages/microservices/exceptions/index.ts

@@ -0,0 +1,17 @@
+import { RpcException } from './rpc-exception';
+
+/**
+ * Exception that instructs Kafka driver to instead of introspecting
+ * error processing flow and sending serialized error message to the consumer,
+ * force bubble it up to the "eachMessage" callback of the underlying "kafkajs" package
+ * (even if interceptors are applied, or an observable stream is returned from the message handler).
+ *
+ * A transient exception that if retried may succeed.
+ *
+ * @publicApi
+ */
+export class KafkaRetriableException extends RpcException {
+  public getError(): string | object {
+    return this;
+  }
+}

packages/microservices/exceptions/kafka-retriable-exception.ts

@@ -13,10 +13,8 @@ import { PipesContextCreator } from '@nestjs/core/pipes/pipes-context-creator';
 import { ClientProxyFactory } from './client';
 import { ClientsContainer } from './container';
 import { ExceptionFiltersContext } from './context/exception-filters-context';
-import { KafkaRpcProxy } from './context/kafka-rpc-proxy';
 import { RpcContextCreator } from './context/rpc-context-creator';
 import { RpcProxy } from './context/rpc-proxy';
-import { Transport } from './enums';
 import { CustomTransportStrategy } from './interfaces';
 import { ListenersController } from './listeners-controller';
 import { Server } from './server/server';
@@ -25,19 +23,13 @@ export class MicroservicesModule {
   private readonly clientsContainer = new ClientsContainer();
   private listenersController: ListenersController;
 
-  public register(
-    container: NestContainer,
-    config: ApplicationConfig,
-    transport?: Transport,
-  ) {
-    const rpcProxy =
-      transport === Transport.KAFKA ? new KafkaRpcProxy() : new RpcProxy();
+  public register(container: NestContainer, config: ApplicationConfig) {
     const exceptionFiltersContext = new ExceptionFiltersContext(
       container,
       config,
     );
     const contextCreator = new RpcContextCreator(
-      rpcProxy,
+      new RpcProxy(),
       exceptionFiltersContext,
       new PipesContextCreator(container, config),
       new PipesConsumer(),

packages/microservices/microservices-module.ts

@@ -46,12 +46,7 @@ export class NestMicroservice
   ) {
     super(container);
 
-    this.microservicesModule.register(
-      container,
-      this.applicationConfig,
-      // @ts-expect-error transport does not exist on type of config??
-      config.transport,
-    );
+    this.microservicesModule.register(container, this.applicationConfig);
     this.createServer(config);
     this.selectContextModule();
   }

packages/microservices/nest-microservice.ts

@@ -1,5 +1,6 @@
 import { Logger } from '@nestjs/common/services/logger.service';
 import { isNil } from '@nestjs/common/utils/shared.utils';
+import { concat, firstValueFrom, of, skip, throwError } from 'rxjs';
 import {
   KAFKA_DEFAULT_BROKER,
   KAFKA_DEFAULT_CLIENT,
@@ -8,6 +9,7 @@ import {
 } from '../constants';
 import { KafkaContext } from '../ctx-host';
 import { KafkaHeaders, Transport } from '../enums';
+import { KafkaRetriableException } from '../exceptions';
 import {
   BrokersFunction,
   Consumer,
@@ -182,10 +184,27 @@ export class ServerKafka extends Server implements CustomTransportStrategy {
       });
     }
 
+    // @TODO pass packet.data.value (overide deserializer)
     const response$ = this.transformToObservable(
       await handler(packet.data, kafkaContext),
     );
-    response$ && this.send(response$, publish);
+
+    const [isError, valueOrError] = await firstValueFrom(response$)
+      .then(value => [false, value])
+      .catch(error => {
+        if (error instanceof KafkaRetriableException) {
+          throw error;
+        }
+        return [true, error];
+      });
+
+    this.send(
+      concat(
+        isError ? throwError(() => valueOrError) : of(valueOrError),
+        response$.pipe(skip(1)),
+      ),
+      publish,
+    );
   }
 
   public async sendMessage(
@@ -227,9 +246,12 @@ export class ServerKafka extends Server implements CustomTransportStrategy {
     if (!outgoingResponse.err) {
       return;
     }
-    outgoingMessage.headers[KafkaHeaders.NEST_ERR] = Buffer.from(
-      outgoingResponse.err,
-    );
+    const stringifiedError =
+      typeof outgoingResponse.err === 'object'
+        ? JSON.stringify(outgoingResponse.err)
+        : outgoingResponse.err;
+    outgoingMessage.headers[KafkaHeaders.NEST_ERR] =
+      Buffer.from(stringifiedError);
   }
 
   public assignCorrelationIdHeader(