Merge pull request #9116 from alphaprinz/notif_backports

[Backport into 5.18] pr 9111, dfs 2834

Author: nimrod-becker

package-lock.json

@@ -107,7 +107,7 @@
     "nan": "2.22.2",
     "ncp": "2.0.0",
     "node-addon-api": "8.3.1",
-    "node-rdkafka": "3.4.0",
+    "node-rdkafka": "3.4.1",
     "performance-now": "2.1.0",
     "pg": "8.16.0",
     "ping": "0.4.4",

package.json

@@ -38,7 +38,7 @@ commit_epoch=$(git show -s --format=%ci ${CEPH_TESTS_VERSION} | awk '{print $1}'
 commit_date=$(date -d ${commit_epoch} +%s)
 current_date=$(date +%s)
 
-max_days="450"
+max_days="600"
 if [ $((current_date-commit_date)) -gt $((3600*24*${max_days})) ]
 then
     echo "ceph tests were not updated for ${max_days} days, Exiting"

src/test/system_tests/ceph_s3_tests/test_ceph_s3_deploy.sh

@@ -103,10 +103,11 @@ class Notificator {
                 throw err;
             } finally {
                 await log.close();
-                this.notif_to_connect.clear();
                 for (const conn of this.connect_str_to_connection.values()) {
                     conn.destroy();
                 }
+                this.connect_str_to_connection.clear();
+                this.notif_to_connect.clear();
             }
         }
     }
@@ -259,18 +260,25 @@ class KafkaNotificator {
 
     async connect() {
         this.connection = new Kafka.HighLevelProducer(this.connect_obj.kafka_options_object);
+        dbg.log2("Kafka producer connecting, connect =", this.connect_obj);
         await new Promise((res, rej) => {
             this.connection.on('ready', () => {
+                dbg.log2("Kafka producer connected for connection =", this.connect_obj);
                 res();
             });
             this.connection.on('connection.failure', err => {
+                dbg.error("Kafka producer failed to connect. connect = ", this.connect_obj, ", err =", err);
                 rej(err);
             });
             this.connection.on('event.log', arg => {
-                dbg.log1("event log", arg);
+                dbg.log2("event log", arg);
+            });
+            this.connection.on('event.error', arg => {
+                dbg.error("event error =", arg);
             });
             this.connection.connect();
         });
+        dbg.log2("Kafka producer's connect done, connect =", this.connect_obj);
         this.connection.setPollInterval(100);
     }
 
@@ -283,10 +291,12 @@ class KafkaNotificator {
                 Buffer.from(JSON.stringify(notif.notif)),
                 null,
                 Date.now(),
-                (err, offset) => {
+                err => {
                     if (err) {
+                        dbg.error("Failed to notify. Connect =", connect_obj, ", notif =", notif);
                         promise_failure_cb(notif, failure_ctxt, err).then(resolve);
                     } else {
+                        dbg.log2("Kafka notify successful. Connect =", connect_obj, ", notif =", notif);
                         resolve();
                     }
                 }
@@ -295,8 +305,10 @@ class KafkaNotificator {
     }
 
     destroy() {
-        this.connection.flush(10000);
-        this.connection.disconnect();
+        if (this.connection.isConnected()) {
+            this.connection.flush(10000);
+            this.connection.disconnect();
+        }
     }
 }