Add persistent event listeners to monitor connection health for both producer and consumer

Author: daniyaltalatteo

src/index.js

@@ -107,6 +107,48 @@ class KafkaClient {
     );
 
     this.#registry = new SchemaRegistry({ host: this.#avroSchemaRegistry });
+
+    // Set up persistent event listeners to monitor connection health for both producer and consumer
+    this.#registerProducerEventListeners();
+    this.#registerConsumerEventListeners();
+  }
+
+  #registerProducerEventListeners() {
+    // If the producer disconnects unexpectedly, set the isProducerConnected flag to false
+    this.#producer.on('disconnected', () => {
+      if (this.#isProducerConnected) {
+        console.error('Kafka producer disconnected unexpectedly');
+        this.#isProducerConnected = false;
+      }
+    });
+
+    // Capture event errors and set the isProducerConnected flag to false in case of an event error
+    this.#producer.on('event.error', (error) => {
+      console.error(`Kafka producer encountered event error: ${error}`);
+      this.#isProducerConnected = false;
+    });
+  }
+
+  #registerConsumerEventListeners() {
+    // If the consumer disconnects unexpectedly, set the isConsumerConnected flag to false
+    this.#consumer.on('disconnected', () => {
+      if (this.#isConsumerConnected) {
+        console.error('Kafka consumer disconnected unexpectedly');
+        
+        this.#isConsumerConnected = false;
+        clearInterval(this.#intervalId);
+        this.#intervalId = null;
+      }
+    });
+
+    // Capture event errors and set the isConsumerConnected flag to false in case of an event error
+    this.#consumer.on('event.error', (error) => {
+      console.error(`Kafka consumer encountered event error: ${error}`);
+      
+      this.#isConsumerConnected = false;
+      clearInterval(this.#intervalId);
+      this.#intervalId = null;
+    });
   }
 
   /**
@@ -119,7 +161,6 @@ class KafkaClient {
         return new Promise((resolve, reject) => {
           // Remove any previously attached listeners for these events
           this.#producer.removeAllListeners('ready');
-          this.#producer.removeAllListeners('event.error');
           this.#producer.removeAllListeners('connection.failure');
 
           this.#producer.connect();
@@ -129,24 +170,17 @@ class KafkaClient {
             console.log('Kafka producer successfully connected');
 
             // Once producer is connected, remove error listeners to avoid handling late errors
-            this.#producer.removeAllListeners('event.error');
             this.#producer.removeAllListeners('connection.failure');
 
             resolve();
           });
 
-          this.#producer.once('event.error', (err) => {
-            this.#isProducerConnected = false;
-            console.error(`Kafka producer connection error: ${err}`);
-            reject(err);
-          });
-
-          this.#producer.once('connection.failure', (err) => {
+          this.#producer.once('connection.failure', (error) => {
             this.#isProducerConnected = false;
             console.error(
-              `Kafka producer connection resulted in failure: ${err}`,
+              `Kafka producer connection resulted in failure: ${error}`,
             );
-            reject(err);
+            reject(error);
           });
         });
       }, retryOptions);
@@ -165,7 +199,6 @@ class KafkaClient {
         return new Promise((resolve, reject) => {
           // Remove any previously attached listeners for these events
           this.#consumer.removeAllListeners('ready');
-          this.#consumer.removeAllListeners('event.error');
           this.#consumer.removeAllListeners('connection.failure');
 
           this.#consumer.connect();
@@ -175,24 +208,17 @@ class KafkaClient {
             console.log('Kafka consumer successfully connected');
 
             // Once consumer is connected, remove error listeners to avoid handling late errors
-            this.#consumer.removeAllListeners('event.error');
             this.#consumer.removeAllListeners('connection.failure');
 
             resolve();
           });
 
-          this.#consumer.once('event.error', (err) => {
-            this.#isConsumerConnected = false;
-            console.error(`Kafka consumer connection error: ${err}`);
-            reject(err);
-          });
-
-          this.#consumer.once('connection.failure', (err) => {
+          this.#consumer.once('connection.failure', (error) => {
             this.#isConsumerConnected = false;
             console.error(
-              `Kafka consumer connection resulted in failure: ${err}`,
+              `Kafka consumer connection resulted in failure: ${error}`,
             );
-            reject(err);
+            reject(error);
           });
         });
       }, retryOptions);
@@ -213,7 +239,7 @@ class KafkaClient {
       }
     } catch (error) {
       console.error(
-        `Error occurred connecting to Kafka producer: ${error.message}`,
+        `Kafka producer connection failed with error ${error.message}. Max retries reached. Exiting...`,
       );
       process.exit(1);
     }
@@ -231,7 +257,7 @@ class KafkaClient {
       }
     } catch (error) {
       console.error(
-        `Error occurred connecting to Kafka consumer: ${error.message}`,
+        `Kafka consumer connection failed with error ${error.message}. Max retries reached. Exiting...`,
       );
       process.exit(1);
     }
@@ -321,21 +347,28 @@ class KafkaClient {
   async disconnectProducer() {
     try {
       if (this.#isProducerConnected) {
-        return new Promise((resolve) => {
-          this.#producer.disconnect();
-
-          this.#producer.once('disconnected', () => {
-            this.#isProducerConnected = false;
-            this.#producer.removeAllListeners();
-
-            console.log('Successfully disconnected Kafka producer');
-            resolve();
+        await backOff(() => {
+          return new Promise((resolve, reject) => {
+            this.#producer.disconnect((error) => {
+              console.error(`Error occurred disconnecting producer: ${error}`);
+              reject(error);
+            });
+
+            this.#producer.once('disconnected', () => {
+              this.#isProducerConnected = false;
+              this.#producer.removeAllListeners();
+
+              console.log('Successfully disconnected Kafka producer');
+              resolve();
+            });
           });
-        });
+        }, retryOptions);
       }
     } catch (error) {
-      console.error(`Error disconnecting Kafka producer: ${error}`);
-      throw new Error(`Error disconnecting Kafka producer: ${error}`);
+      console.error(
+        `Kafka producer disconnection failed with error ${error.message}. Max retries reached. Exiting...`,
+      );
+      process.exit(1);
     }
   }
 
@@ -346,26 +379,34 @@ class KafkaClient {
   async disconnectConsumer() {
     try {
       if (this.#isConsumerConnected) {
-        return new Promise((resolve) => {
-          this.#consumer.disconnect();
-
-          this.#consumer.once('disconnected', () => {
-            this.#isConsumerConnected = false;
-            this.#consumer.removeAllListeners();
-
-            clearInterval(this.#intervalId);
-            this.#intervalId = null;
-
-            console.log('Successfully disconnected Kafka consumer');
-            resolve();
+        await backOff(() => {
+          return new Promise((resolve, reject) => {
+            this.#consumer.disconnect((error) => {
+              console.error(`Error occurred disconnecting consumer: ${error}`);
+              reject(error);
+            });
+
+            this.#consumer.once('disconnected', () => {
+              this.#isConsumerConnected = false;
+              this.#consumer.removeAllListeners();
+
+              clearInterval(this.#intervalId);
+              this.#intervalId = null;
+
+              console.log('Successfully disconnected Kafka consumer');
+              resolve();
+            });
           });
-        });
+        }, retryOptions);
       }
     } catch (error) {
-      console.error(`Error disconnecting Kafka consumer: ${error}`);
       clearInterval(this.#intervalId);
       this.#intervalId = null;
-      throw new Error(`Error disconnecting Kafka consumer: ${error}`);
+
+      console.error(
+        `Kafka consumer disconnection failed with error ${error.message}. Max retries reached. Exiting...`,
+      );
+      process.exit(1);
     }
   }
 }