Stats callback

Author: emasab

lib/kafkajs/_admin.js

@@ -174,6 +174,12 @@ class Admin {
     /* Delete properties which are already processed, or cannot be passed to node-rdkafka */
     delete rdKafkaConfig.kafkaJS;
 
+    if (Object.hasOwn(rdKafkaConfig, 'stats_cb')) {
+      if (typeof rdKafkaConfig['stats_cb'] === 'function')
+        this.#statsCb = rdKafkaConfig['stats_cb'];
+      delete rdKafkaConfig['stats_cb'];
+    }
+
     return rdKafkaConfig;
   }
 
@@ -203,6 +209,12 @@ class Admin {
     }
   }
 
+  /**
+   * Callback for the event.stats event, if defined.
+   * @private
+   */
+  #statsCb = null;
+
   /**
    * Set up the client and connect to the bootstrap brokers.
    * @returns {Promise<void>} Resolves when connection is complete, rejects on error.
@@ -225,6 +237,9 @@ class Admin {
             'ready': this.#readyCb.bind(this),
             'event.log': (msg) => loggerTrampoline(msg, this.#logger),
           });
+          if (this.#statsCb) {
+               this.#internalClient.on('event.stats',  this.#statsCb.bind(this));
+          }
         } else {
           const underlyingClient = this.#existingClient._getInternalClient();
           if (!underlyingClient) {

lib/kafkajs/_consumer.js

@@ -731,6 +731,12 @@ class Consumer {
       delete rdKafkaConfig['js.consumer.max.cache.size.per.worker.ms'];
     }
 
+    if (Object.hasOwn(rdKafkaConfig, 'stats_cb')) {
+      if (typeof rdKafkaConfig['stats_cb'] === 'function')
+        this.#statsCb = rdKafkaConfig['stats_cb'];
+      delete rdKafkaConfig['stats_cb'];
+    }
+
     return rdKafkaConfig;
   }
 
@@ -770,6 +776,13 @@ class Consumer {
     }
   }
 
+
+  /**
+   * Callback for the event.stats event, if defined.
+   * @private
+   */
+  #statsCb = null;
+
   /**
    * Converts headers returned by node-rdkafka into a format that can be used by the eachMessage/eachBatch callback.
    * @param {import("../..").MessageHeader[] | undefined} messageHeaders
@@ -1167,6 +1180,9 @@ class Consumer {
     this.#internalClient.on('error', this.#errorCb.bind(this));
     this.#internalClient.on('event.error', this.#errorCb.bind(this));
     this.#internalClient.on('event.log', (msg) => loggerTrampoline(msg, this.#logger));
+    if (this.#statsCb) {
+      this.#internalClient.on('event.stats', this.#statsCb.bind(this));
+    }
 
     return new Promise((resolve, reject) => {
       this.#connectPromiseFunc = { resolve, reject };

lib/kafkajs/_producer.js

@@ -269,6 +269,12 @@ class Producer {
     /* TODO: Add a warning if dr_cb is set? Or else, create a trampoline for it. */
     rdKafkaConfig.dr_cb = true;
 
+    if (Object.hasOwn(rdKafkaConfig, 'stats_cb')) {
+      if (typeof rdKafkaConfig['stats_cb'] === 'function')
+        this.#statsCb = rdKafkaConfig['stats_cb'];
+      delete rdKafkaConfig['stats_cb'];
+    }
+
     return rdKafkaConfig;
   }
 
@@ -373,6 +379,12 @@ class Producer {
     this.#logger.error(err, this.#createProducerBindingMessageMetadata());
   }
 
+  /**
+   * Callback for the event.stats event, if defined.
+   * @private
+   */
+  #statsCb = null;
+
   /**
    * Set up the client and connect to the bootstrap brokers.
    *
@@ -394,6 +406,9 @@ class Producer {
     this.#internalClient.on('event.error', this.#errorCb.bind(this));
     this.#internalClient.on('error', this.#errorCb.bind(this));
     this.#internalClient.on('event.log', (msg) => loggerTrampoline(msg, this.#logger));
+    if (this.#statsCb) {
+      this.#internalClient.on('event.stats', this.#statsCb.bind(this));
+    }
 
     return new Promise((resolve, reject) => {
       this.#connectPromiseFunc = { resolve, reject };