From 8e27fd19253761b2df477c80ecc35eb1a7d66b49 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Jo=C3=A3o=20Filipe=20Silva?=
 <41396279+jfgsilva@users.noreply.github.com>
Date: Thu, 4 Jun 2026 11:36:40 +0100
Subject: [PATCH] fix: add timeout functionality to kafka (#7472)

Co-authored-by: joaofilipesilva <joao.silva@clearops.com>
---
 server/model/monitor.js                           |  1 +
 server/util-server.js                             |  9 ++++++++-
 src/lang/en.json                                  |  1 +
 src/pages/EditMonitor.vue                         | 13 +++++++++++--
 test/backend-test/monitors/test-kafka-producer.js | 15 +++++++++++++++
 5 files changed, 36 insertions(+), 3 deletions(-)
 create mode 100644 test/backend-test/monitors/test-kafka-producer.js

diff --git a/server/model/monitor.js b/server/model/monitor.js
index 4df6d0e9e..8e1843cd6 100644
--- a/server/model/monitor.js
+++ b/server/model/monitor.js
@@ -939,6 +939,7 @@ class Monitor extends BeanModel {
                             ssl: this.kafkaProducerSsl,
                             clientId: `Uptime-Kuma/${version}`,
                             interval: this.interval,
+                            connectionTimeout: this.timeout,
                         },
                         JSON.parse(this.kafkaProducerSaslOptions)
                     );
diff --git a/server/util-server.js b/server/util-server.js
index 23de51b12..a6bf06119 100644
--- a/server/util-server.js
+++ b/server/util-server.js
@@ -213,7 +213,13 @@ exports.pingAsync = function (
  */
 exports.kafkaProducerAsync = function (brokers, topic, message, options = {}, saslOptions = {}) {
     return new Promise((resolve, reject) => {
-        const { interval = 20, allowAutoTopicCreation = false, ssl = false, clientId = "Uptime-Kuma" } = options;
+        const {
+            interval = 20,
+            allowAutoTopicCreation = false,
+            ssl = false,
+            clientId = "Uptime-Kuma",
+            connectionTimeout = 1,
+        } = options;
 
         let connectedToKafka = false;
 
@@ -238,6 +244,7 @@ exports.kafkaProducerAsync = function (brokers, topic, message, options = {}, sa
                 retries: 0,
             },
             ssl: ssl,
+            connectionTimeout: connectionTimeout * 1000,
         });
 
         let producer = client.producer({
diff --git a/src/lang/en.json b/src/lang/en.json
index a7dbc97c7..c6c1259c6 100644
--- a/src/lang/en.json
+++ b/src/lang/en.json
@@ -87,6 +87,7 @@
     "Path": "Path",
     "Heartbeat Interval": "Heartbeat Interval",
     "Request Timeout": "Request Timeout",
+    "Connection Timeout": "Connection Timeout",
     "timeoutAfter": "Timeout after {0} seconds",
     "Retries": "Retries",
     "Heartbeat Retry Interval": "Heartbeat Retry Interval",
diff --git a/src/pages/EditMonitor.vue b/src/pages/EditMonitor.vue
index 2ee2e66ae..b93d0a928 100644
--- a/src/pages/EditMonitor.vue
+++ b/src/pages/EditMonitor.vue
@@ -1499,12 +1499,19 @@
                                     monitor.type === 'ping' ||
                                     monitor.type === 'rabbitmq' ||
                                     monitor.type === 'snmp' ||
-                                    monitor.type === 'websocket-upgrade'
+                                    monitor.type === 'websocket-upgrade' ||
+                                    monitor.type === 'kafka-producer'
                                 "
                                 class="my-3"
                             >
                                 <label for="timeout" class="form-label">
-                                    {{ monitor.type === "ping" ? $t("pingGlobalTimeoutLabel") : $t("Request Timeout") }}
+                                    {{
+                                        monitor.type === "ping"
+                                            ? $t("pingGlobalTimeoutLabel")
+                                            : monitor.type === "kafka-producer"
+                                              ? $t("Connection Timeout")
+                                              : $t("Request Timeout")
+                                    }}
                                     <span v-if="monitor.type !== 'ping'">
                                         ({{ $t("timeoutAfter", [monitor.timeout || clampTimeout(monitor.interval)]) }})
                                     </span>
@@ -3659,6 +3666,8 @@ message HealthCheckResponse {
                     this.monitor.timeout = 5;
                 } else if (this.monitor.type === "ping") {
                     this.monitor.timeout = 10;
+                } else if (this.monitor.type === "kafka-producer") {
+                    this.monitor.timeout = 1;
                 } else {
                     this.monitor.timeout = 48;
                 }
diff --git a/test/backend-test/monitors/test-kafka-producer.js b/test/backend-test/monitors/test-kafka-producer.js
new file mode 100644
index 000000000..ec8dd5022
--- /dev/null
+++ b/test/backend-test/monitors/test-kafka-producer.js
@@ -0,0 +1,15 @@
+const { describe, test } = require("node:test");
+const assert = require("node:assert");
+const { kafkaProducerAsync } = require("../../../server/util-server");
+
+describe("Kafka Producer", () => {
+    test("rejects when broker is not reachable", async () => {
+        await assert.rejects(
+            kafkaProducerAsync(["localhost:19092"], "test-topic", "test-message", {
+                interval: 5,
+                connectionTimeout: 1,
+            }),
+            /.*/ // any error
+        );
+    });
+});