fix(microservices): messages emitted before kafka client is ready fail

packages/microservices/client/client-kafka.ts

@@ -42,13 +42,13 @@ let kafkaPackage: any = {};
 
 export class ClientKafka extends ClientProxy {
   protected logger = new Logger(ClientKafka.name);
-  protected client: Kafka = null;
-  protected consumer: Consumer = null;
-  protected producer: Producer = null;
-  protected parser: KafkaParser = null;
+  protected client: Kafka | null = null;
+  protected consumer: Consumer | null = null;
+  protected producer: Producer | null = null;
+  protected parser: KafkaParser | null = null;
+  protected initialized: Promise<void> | null = null;
   protected responsePatterns: string[] = [];
   protected consumerAssignments: { [key: string]: number } = {};
-
   protected brokers: string[] | BrokersFunction;
   protected clientId: string;
   protected groupId: string;
@@ -94,46 +94,56 @@ export class ClientKafka extends ClientProxy {
     this.consumer && (await this.consumer.disconnect());
     this.producer = null;
     this.consumer = null;
+    this.initialized = null;
     this.client = null;
   }
 
   public async connect(): Promise<Producer> {
-    if (this.client) {
-      return this.producer;
+    if (this.initialized) {
+      return this.initialized.then(() => this.producer);
     }
-    this.client = this.createClient();
-
-    if (!this.producerOnlyMode) {
-      const partitionAssigners = [
-        (
-          config: ConstructorParameters<typeof KafkaReplyPartitionAssigner>[1],
-        ) => new KafkaReplyPartitionAssigner(this, config),
-      ] as any[];
-
-      const consumerOptions = Object.assign(
-        {
-          partitionAssigners,
-        },
-        this.options.consumer || {},
-        {
-          groupId: this.groupId,
-        },
-      );
+    this.initialized = new Promise(async (resolve, reject) => {
+      try {
+        this.client = this.createClient();
+
+        if (!this.producerOnlyMode) {
+          const partitionAssigners = [
+            (
+              config: ConstructorParameters<
+                typeof KafkaReplyPartitionAssigner
+              >[1],
+            ) => new KafkaReplyPartitionAssigner(this, config),
+          ];
+
+          const consumerOptions = Object.assign(
+            {
+              partitionAssigners,
+            },
+            this.options.consumer || {},
+            {
+              groupId: this.groupId,
+            },
+          );
 
-      this.consumer = this.client.consumer(consumerOptions);
-      // set member assignments on join and rebalance
-      this.consumer.on(
-        this.consumer.events.GROUP_JOIN,
-        this.setConsumerAssignments.bind(this),
-      );
-      await this.consumer.connect();
-      await this.bindTopics();
-    }
+          this.consumer = this.client.consumer(consumerOptions);
+          // set member assignments on join and rebalance
+          this.consumer.on(
+            this.consumer.events.GROUP_JOIN,
+            this.setConsumerAssignments.bind(this),
+          );
+          await this.consumer.connect();
+          await this.bindTopics();
+        }
 
-    this.producer = this.client.producer(this.options.producer || {});
-    await this.producer.connect();
+        this.producer = this.client.producer(this.options.producer || {});
+        await this.producer.connect();
 
-    return this.producer;
+        resolve();
+      } catch (err) {
+        reject(err);
+      }
+    });
+    return this.initialized.then(() => this.producer);
   }
 
   public async bindTopics(): Promise<void> {

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

@@ -11,7 +11,6 @@ import {
 } from '../../external/kafka.interface';
 
 describe('ClientKafka', () => {
-  // static
   const topic = 'test.topic';
   const partition = 0;
   const replyTopic = 'test.topic.reply';
@@ -25,7 +24,6 @@ describe('ClientKafka', () => {
   const heartbeat = async () => {};
   const pause = () => () => {};
 
-  // message
   const message: KafkaMessage = {
     key: Buffer.from(key),
     offset,
@@ -35,7 +33,6 @@ describe('ClientKafka', () => {
     attributes,
   };
 
-  // deserialized message
   const deserializedMessage: any = {
     key,
     offset,
@@ -47,7 +44,6 @@ describe('ClientKafka', () => {
     partition,
   };
 
-  // payloads
   const payload: EachMessagePayload = {
     topic,
     partition,
@@ -152,26 +148,6 @@ describe('ClientKafka', () => {
     pause,
   };
 
-  const deserializedPayloadError: EachMessagePayload = {
-    topic,
-    partition,
-    message: Object.assign(
-      {
-        headers: {
-          [KafkaHeaders.CORRELATION_ID]: correlationId,
-          [KafkaHeaders.NEST_ERR]: NO_MESSAGE_HANDLER,
-        },
-      },
-      deserializedMessage,
-      {
-        size: 0,
-        value: null,
-      },
-    ),
-    heartbeat,
-    pause,
-  };
-
   let client: ClientKafka;
   let callback: sinon.SinonSpy;
   let connect: sinon.SinonSpy;
@@ -182,7 +158,7 @@ describe('ClientKafka', () => {
   let consumerStub: sinon.SinonStub;
   let producerStub: sinon.SinonStub;
   let createClientStub: sinon.SinonStub;
-  let kafkaClient;
+  let kafkaClient: any;
 
   beforeEach(() => {
     client = new ClientKafka({});
@@ -322,7 +298,8 @@ describe('ClientKafka', () => {
       });
 
       it('should expect the connection to be reused', async () => {
-        (client as any).client = kafkaClient;
+        (client as any).initialized = Promise.resolve({});
+
         await client.connect();
 
         expect(createClientStub.calledOnce).to.be.false;
@@ -368,7 +345,8 @@ describe('ClientKafka', () => {
       });
 
       it('should expect the connection to be reused', async () => {
-        (client as any).client = kafkaClient;
+        (client as any).initialized = Promise.resolve({});
+
         await client.connect();
 
         expect(createClientStub.calledOnce).to.be.false;