fix(microservice): connect to kafka with producerOnlyMode per default

resolves: #1690
nestjs · Sep 14, 2023 · 3e44d7a · 3e44d7a
1 parent bc6bec1
commit 3e44d7a
Show file tree

Hide file tree

Showing 6 changed files with 191 additions and 75 deletions.
diff --git a/docker-compose.yml b/docker-compose.yml
@@ -21,13 +21,43 @@ services:
     ports:
       - 27017:27017
 
-  redis_test:
+  rabbitmq:
+    image: rabbitmq:3-management-alpine
+    container_name: 'rabbitmq'
+    networks:
+        - overlay
+    ports:
+    - 5672:5672
+    - 15672:15672
+
+  redis:
     image: redis:latest
-    hostname: redis_test
     networks:
       - overlay
     ports:
       - 6379:6379
 
+  kafka_zookeeper:
+    image: confluentinc/cp-zookeeper:7.0.1
+    container_name: kafka_zookeeper
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+
+  kafka_broker:
+    image: confluentinc/cp-kafka:7.0.1
+    container_name: broker
+    ports:
+      - "9092:9092"
+    depends_on:
+      - kafka_zookeeper
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: 'kafka_zookeeper:2181'
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,PLAINTEXT_INTERNAL://kafka_broker:29092
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
 networks:
   overlay:
diff --git a/e2e/health-checks/microservice.health.e2e-spec.ts b/e2e/health-checks/microservice.health.e2e-spec.ts
@@ -5,7 +5,12 @@ import {
   bootstrapTestingModule,
   DynamicHealthEndpointFn,
 } from '../helper';
-import { Transport } from '@nestjs/microservices';
+import {
+  KafkaOptions,
+  RmqOptions,
+  TcpClientOptions,
+  Transport,
+} from '@nestjs/microservices';
 
 describe('MicroserviceHealthIndicator', () => {
   let app: INestApplication;
@@ -20,80 +25,131 @@ describe('MicroserviceHealthIndicator', () => {
     microservice = await bootstrapMicroservice();
   });
 
-  it('should check if the microservice is available', async () => {
-    app = await setHealthEndpoint(({ healthCheck, microservice }) =>
-      healthCheck.check([
-        async () =>
-          microservice.pingCheck('tcp', {
-            transport: Transport.TCP,
-            options: {
-              host: '0.0.0.0',
-              port: 8889,
-            },
-          }),
-      ]),
-    ).start();
-
-    const details = { tcp: { status: 'up' } };
-    return request(app.getHttpServer()).get('/health').expect(200).expect({
-      status: 'ok',
-      info: details,
-      error: {},
-      details,
+  describe('TCP', () => {
+    it('should connect', async () => {
+      app = await setHealthEndpoint(({ healthCheck, microservice }) =>
+        healthCheck.check([
+          async () =>
+            microservice.pingCheck('tcp', {
+              transport: Transport.TCP,
+              options: {
+                host: '0.0.0.0',
+                port: 8889,
+              },
+            }),
+        ]),
+      ).start();
+
+      const details = { tcp: { status: 'up' } };
+      return request(app.getHttpServer()).get('/health').expect(200).expect({
+        status: 'ok',
+        info: details,
+        error: {},
+        details,
+      });
+    });
+
+    it('should throw an error if is not reachable', async () => {
+      app = await setHealthEndpoint(({ healthCheck, microservice }) =>
+        healthCheck.check([
+          async () =>
+            microservice.pingCheck<TcpClientOptions>('tcp', {
+              transport: Transport.TCP,
+              options: {
+                host: '0.0.0.0',
+                port: 8889,
+              },
+            }),
+        ]),
+      ).start();
+
+      await microservice.close();
+
+      const details = {
+        tcp: { status: 'down', message: 'connect ECONNREFUSED 0.0.0.0:8889' },
+      };
+      return request(app.getHttpServer()).get('/health').expect(503).expect({
+        status: 'error',
+        info: {},
+        error: details,
+        details,
+      });
     });
   });
 
-  it('should throw an error if the service is not reachable', async () => {
-    app = await setHealthEndpoint(({ healthCheck, microservice }) =>
-      healthCheck.check([
-        async () =>
-          microservice.pingCheck('tcp', {
-            transport: Transport.TCP,
-            options: {
-              host: '0.0.0.0',
-              port: 8889,
-            },
-          }),
-      ]),
-    ).start();
-
-    await microservice.close();
-
-    const details = {
-      tcp: { status: 'down', message: 'connect ECONNREFUSED 0.0.0.0:8889' },
-    };
-    return request(app.getHttpServer()).get('/health').expect(503).expect({
-      status: 'error',
-      info: {},
-      error: details,
-      details,
+  describe('RMQ', () => {
+    it('should connect', async () => {
+      app = await setHealthEndpoint(({ healthCheck, microservice }) =>
+        healthCheck.check([
+          async () =>
+            microservice.pingCheck<RmqOptions>('rmq', {
+              transport: Transport.RMQ,
+              options: {
+                urls: ['amqp://localhost:5672'],
+              },
+            }),
+        ]),
+      ).start();
+
+      const details = { rmq: { status: 'up' } };
+      return request(app.getHttpServer()).get('/health').expect(200).expect({
+        status: 'ok',
+        info: details,
+        error: {},
+        details,
+      });
+    });
+
+    it('should throw an error if is not reachable', async () => {
+      app = await setHealthEndpoint(({ healthCheck, microservice }) =>
+        healthCheck.check([
+          async () =>
+            microservice.pingCheck<RmqOptions>('rmq', {
+              transport: Transport.RMQ,
+              options: {
+                urls: ['amqp://0.0.0.0:8889'],
+              },
+            }),
+        ]),
+      ).start();
+
+      await microservice.close();
+
+      const details = {
+        rmq: { status: 'down', message: 'rmq is not available' },
+      };
+      return request(app.getHttpServer()).get('/health').expect(503).expect({
+        status: 'error',
+        info: {},
+        error: details,
+        details,
+      });
     });
   });
 
-  it('should throw an error if an RMQ microservice is not reachable', async () => {
-    app = await setHealthEndpoint(({ healthCheck, microservice }) =>
-      healthCheck.check([
-        async () =>
-          microservice.pingCheck('rmq', {
-            transport: Transport.RMQ,
-            options: {
-              host: '0.0.0.0',
-              port: 8889,
-            },
-          }),
-      ]),
-    ).start();
-
-    await microservice.close();
-
-    const details = {
-      rmq: { status: 'down', message: 'rmq is not available' },
-    };
-    return request(app.getHttpServer()).get('/health').expect(503).expect({
-      status: 'error',
-      info: {},
-      error: details,
-      details,
+  describe('Kafka', () => {
+    it('should connect', async () => {
+      app = await setHealthEndpoint(({ healthCheck, microservice }) =>
+        healthCheck.check([
+          async () =>
+            microservice.pingCheck<KafkaOptions>('kafka', {
+              transport: Transport.KAFKA,
+              options: {
+                client: {
+                  brokers: ['localhost:9092'],
+                },
+              },
+            }),
+        ]),
+      ).start();
+
+      const details = { kafka: { status: 'up' } };
+      return request(app.getHttpServer()).get('/health').expect(200).expect({
+        status: 'ok',
+        info: details,
+        error: {},
+        details,
+      });
     });
   });
 

diff --git a/lib/health-indicator/microservice/microservice.health.ts b/lib/health-indicator/microservice/microservice.health.ts
@@ -112,7 +112,7 @@ export class MicroserviceHealthIndicator extends HealthIndicator {
    * @throws {HealthCheckError} If the microservice is not reachable
    *
    * @example
-   * microservice.pingCheck('tcp', {
+   * microservice.pingCheck<TcpClientOptions>('tcp', {
    *   transport: Transport.TCP,
    *   options: { host: 'localhost', port: 3001 },
    * })
@@ -124,6 +124,15 @@ export class MicroserviceHealthIndicator extends HealthIndicator {
     let isHealthy = false;
     const timeout = options.timeout || 1000;
 
+    if (options.transport === NestJSMicroservices.Transport.KAFKA) {
+      options.options = {
+        // We need to set the producerOnlyMode to true in order to speed
+        // up the connection process. https://github.com/nestjs/terminus/issues/1690
+        producerOnlyMode: true,
+        ...options.options,
+      };
+    }
+
     try {
       await promiseTimeout(timeout, this.pingMicroservice(options));
       isHealthy = true;

diff --git a/package-lock.json b/package-lock.json
diff --git a/package.json b/package.json
@@ -83,6 +83,7 @@
     "husky": "8.0.3",
     "ioredis": "5.3.2",
     "jest": "29.7.0",
+    "kafkajs": "^2.2.4",
     "lint-staged": "13.2.3",
     "mongoose": "7.3.3",
     "mysql2": "3.6.1",

diff --git a/sample/002-microservice-app/src/health/health.controller.ts b/sample/002-microservice-app/src/health/health.controller.ts
@@ -1,5 +1,9 @@
 import { Controller, Get } from '@nestjs/common';
-import { Transport } from '@nestjs/microservices';
+import {
+  RedisOptions,
+  TcpClientOptions,
+  Transport,
+} from '@nestjs/microservices';
 import {
   HealthCheck,
   HealthCheckService,
@@ -18,12 +22,12 @@ export class HealthController {
   check() {
     return this.health.check([
       async () =>
-        this.microservice.pingCheck('tcp', {
+        this.microservice.pingCheck<TcpClientOptions>('tcp', {
           transport: Transport.TCP,
           options: { host: 'localhost', port: 8889 },
         }),
       async () =>
-        this.microservice.pingCheck('redis', {
+        this.microservice.pingCheck<RedisOptions>('redis', {
           transport: Transport.REDIS,
           options: {
             host: 'localhost',