🐛 fix: kafka producer bugs (#3771)

* 🐛 fix: missing Kafka Producer SSL option in frontend object Signed-off-by: Muhammed Hussein Karimi <info@karimi.dev> * ♻️ refactor: better error handling of kafka producer Signed-off-by: Muhammed Hussein Karimi <info@karimi.dev> --------- Signed-off-by: Muhammed Hussein Karimi <info@karimi.dev>
louislam · Sep 23, 2023 · 2ab21cc · 2ab21cc
1 parent 90d0e8c
commit 2ab21cc
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 13 deletions.
diff --git a/server/util-server.js b/server/util-server.js
@@ -286,22 +286,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) => {
@@ -313,8 +313,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", (_) => {

diff --git a/src/pages/EditMonitor.vue b/src/pages/EditMonitor.vue
@@ -880,6 +880,7 @@ const monitorDefaults = {
     kafkaProducerSaslOptions: {
         mechanism: "None",
     },
+    kafkaProducerSsl: false,
     gamedigGivenPortOnly: true,
 };