♻️ refactor: better error handling of kafka producer

Signed-off-by: Muhammed Hussein Karimi <info@karimi.dev>
louislam · Sep 19, 2023 · 35f117c · 35f117c
1 parent 14e0c77
commit 35f117c
Showing 1 changed file with 15 additions and 13 deletions.
diff --git a/server/util-server.js b/server/util-server.js
@@ -290,22 +290,22 @@ exports.kafkaProducerAsync = function (brokers, topic, message, options = {}, sa
 
         producer.connect().then(
             () => {
-                try {
-                    producer.send({
-                        topic: topic,
-                        messages: [{
-                            value: message,
-                        }],
-                    });
-                    connectedToKafka = true;
-                    clearTimeout(timeoutID);
+                producer.send({
+                    topic: topic,
+                    messages: [{
+                        value: message,
+                    }],
+                }).then((_) => {
                     resolve("Message sent successfully");
-                } catch (e) {
+                }).catch((e) => {
                     connectedToKafka = true;
                     producer.disconnect();
                     clearTimeout(timeoutID);
                     reject(new Error("Error sending message: " + e.message));
-                }
+                }).finally(() => {
+                    connectedToKafka = true;
+                    clearTimeout(timeoutID);
+                });
             }
         ).catch(
             (e) => {
@@ -317,8 +317,10 @@ exports.kafkaProducerAsync = function (brokers, topic, message, options = {}, sa
         );
 
         producer.on("producer.network.request_timeout", (_) => {
-            clearTimeout(timeoutID);
-            reject(new Error("producer.network.request_timeout"));
+            if (!connectedToKafka) {
+                clearTimeout(timeoutID);
+                reject(new Error("producer.network.request_timeout"));
+            }
         });
 
         producer.on("producer.disconnect", (_) => {