Add binding level debug logging and client name to logs

* Add binding level debug logging and client name to logs

* Add C++ changes for binding-level logging

Author: milindl

lib/kafkajs/_admin.js

@@ -3,9 +3,12 @@ const { kafkaJSToRdKafkaConfig,
   createKafkaJsErrorFromLibRdKafkaError,
   DefaultLogger,
   CompatibilityErrorMessages,
+  createBindingMessageMetadata,
   logLevel,
+  checkAllowedKeys,
   loggerTrampoline,
-  checkAllowedKeys } = require('./_common');
+  severityToLogLevel,
+} = require('./_common');
 const error = require('./_error');
 
 /**
@@ -57,6 +60,20 @@ class Admin {
    */
   #connectPromiseFunc = null;
 
+  /**
+   * The client name used by the admin client for logging - determined by librdkafka
+   * using a combination of clientId and an integer.
+   * @type {string|undefined}
+   */
+  #clientName = undefined;
+
+  /**
+   * Convenience function to create the metadata object needed for logging.
+   */
+  #createAdminBindingMessageMetadata() {
+    return createBindingMessageMetadata(this.#clientName);
+  }
+
   /**
    * @constructor
    * @param {import("../../types/kafkajs").AdminConstructorConfig} config
@@ -82,6 +99,17 @@ class Admin {
     }
 
     const rdKafkaConfig = kafkaJSToRdKafkaConfig(kjsConfig);
+
+    /* Set the logger */
+    if (Object.hasOwn(kjsConfig, 'logger')) {
+      this.#logger = kjsConfig.logger;
+    }
+
+    /* Set the log level - INFO for compatibility with kafkaJS, or DEBUG if that is turned
+    * on using the logLevel property. rdKafkaConfig.log_level is guaranteed to be set if we're
+    * here, and containing the correct value. */
+    this.#logger.setLogLevel(severityToLogLevel[rdKafkaConfig.log_level]);
+
     return rdKafkaConfig;
   }
 
@@ -121,7 +149,7 @@ class Admin {
     if (this.#state === AdminState.CONNECTING) {
       this.#connectPromiseFunc['reject'](err);
     } else {
-      this.#logger.error(err);
+      this.#logger.error(`Error: ${err.message}`, this.#createAdminBindingMessageMetadata());
     }
   }
 
@@ -145,8 +173,11 @@ class Admin {
         this.#internalClient = RdKafka.AdminClient.create(config, {
           'error': this.#errorCb.bind(this),
           'ready': this.#readyCb.bind(this),
-          'event.log': (msg) => loggerTrampoline(msg, this.#logger)
+          'event.log': (msg) => loggerTrampoline(msg, this.#logger),
         });
+
+        this.#clientName = this.#internalClient.name;
+        this.#logger.info("Admin client connected", this.#createAdminBindingMessageMetadata());
       } catch (err) {
         reject(createKafkaJsErrorFromLibRdKafkaError(err));
       }
@@ -174,6 +205,7 @@ class Admin {
         /* AdminClient disconnect for node-rdkakfa is synchronous. */
         this.#internalClient.disconnect();
         this.#state = AdminState.DISCONNECTED;
+        this.#logger.info("Admin client disconnected", this.#createAdminBindingMessageMetadata());
         resolve();
       } catch (err) {
         reject(createKafkaJsErrorFromLibRdKafkaError(err));

lib/kafkajs/_common.js

@@ -113,6 +113,20 @@ class DefaultLogger {
   }
 }
 
+/**
+ * Convenience function to create a new object to be used as metadata for log messages.
+ * Returned object is intended to be used immediately and not stored.
+ *
+ * @param {string|undefined} clientName
+ */
+function createBindingMessageMetadata(clientName) {
+  return {
+    name: clientName,
+    fac: 'BINDING',
+    timestamp: Date.now(),
+  };
+}
+
 /**
  * Trampoline for user defined logger, if any.
  * @param {{severity: number, fac: string, message: string}} msg
@@ -128,16 +142,16 @@ function loggerTrampoline(msg, logger) {
     case logLevel.NOTHING:
       break;
     case logLevel.ERROR:
-      logger.error(msg.message, { fac: msg.fac, timestamp: Date.now() });
+      logger.error(msg.message, { fac: msg.fac, timestamp: Date.now(), name: msg.name });
       break;
     case logLevel.WARN:
-      logger.warn(msg.message, { fac: msg.fac, timestamp: Date.now() });
+      logger.warn(msg.message, { fac: msg.fac, timestamp: Date.now(), name: msg.name });
       break;
     case logLevel.INFO:
-      logger.info(msg.message, { fac: msg.fac, timestamp: Date.now() });
+      logger.info(msg.message, { fac: msg.fac, timestamp: Date.now(), name: msg.name });
       break;
     case logLevel.DEBUG:
-      logger.debug(msg.message, { fac: msg.fac, timestamp: Date.now() });
+      logger.debug(msg.message, { fac: msg.fac, timestamp: Date.now(), name: msg.name });
       break;
     default:
       throw new error.KafkaJSError("Invalid logLevel", {
@@ -706,6 +720,7 @@ module.exports = {
   topicPartitionOffsetMetadataToKafkaJS,
   createKafkaJsErrorFromLibRdKafkaError,
   convertToRdKafkaHeaders,
+  createBindingMessageMetadata,
   notImplemented,
   logLevel,
   loggerTrampoline,

lib/kafkajs/_consumer.js

@@ -6,6 +6,7 @@ const {
   topicPartitionOffsetToRdKafka,
   topicPartitionOffsetMetadataToRdKafka,
   topicPartitionOffsetMetadataToKafkaJS,
+  createBindingMessageMetadata,
   createKafkaJsErrorFromLibRdKafkaError,
   notImplemented,
   loggerTrampoline,
@@ -171,6 +172,20 @@ class Consumer {
    */
   #topicPartitionToBatchPayload = new Map();
 
+  /**
+   * The client name used by the consumer for logging - determined by librdkafka
+   * using a combination of clientId and an integer.
+   * @type {string|undefined}
+   */
+  #clientName = undefined;
+
+  /**
+   * Convenience function to create the metadata object needed for logging.
+   */
+  #createConsumerBindingMessageMetadata() {
+    return createBindingMessageMetadata(this.#clientName);
+  }
+
   /**
    * @constructor
    * @param {import("../../types/kafkajs").ConsumerConfig} kJSConfig
@@ -222,8 +237,8 @@ class Consumer {
     }
 
     /* TODO: we should cry more about this and render the consumer unusable. */
-    await Promise.all(seekPromises).catch(err => this.#logger.error("Seek error. This is effectively a fatal error:" + err));
-
+    await Promise.all(seekPromises)
+      .catch(err => this.#logger.error(`Seek error. This is effectively a fatal error: ${err}`), this.#createConsumerBindingMessageMetadata());
 
     /* Clear the cache and stored offsets.
      * We need to do this only if topicPartitions = null (global cache expiry).
@@ -242,6 +257,9 @@ class Consumer {
   #rebalanceCallback(err, assignment) {
     err = LibrdKafkaError.create(err);
     const userSpecifiedRebalanceCb = this.#userConfig['rebalance_cb'];
+    this.#logger.info(
+      `Received rebalance event with message: '${err.message}' and ${assignment.length} partition(s)`,
+      this.#createConsumerBindingMessageMetadata());
 
     let assignmentFnCalled = false;
     function assignmentFn(userAssignment) {
@@ -559,6 +577,9 @@ class Consumer {
     /* Slight optimization for cases where the size of messages in our subscription is less than the cache size. */
     this.#internalClient.setDefaultIsTimeoutOnlyForFirstMessage(true);
 
+    this.#clientName = this.#internalClient.name;
+    this.#logger.info('Consumer connected', this.#createConsumerBindingMessageMetadata());
+
     // Resolve the promise.
     this.#connectPromiseFunc['resolve']();
   }
@@ -571,7 +592,7 @@ class Consumer {
     if (this.#state === ConsumerState.CONNECTING) {
       this.#connectPromiseFunc['reject'](err);
     } else {
-      this.#logger.error(err);
+      this.#logger.error(err, this.#createConsumerBindingMessageMetadata());
     }
   }
 
@@ -660,8 +681,7 @@ class Consumer {
       this.#lastConsumedOffsets.set(key, offset + 1);
     } catch (e) {
       /* Not much we can do, except log the error. */
-      if (this.#logger)
-        this.#logger.error(`Consumer encountered error while storing offset. Error details: ${e}:${e.stack}`);
+      this.#logger.error(`Consumer encountered error while storing offset. Error details: ${e}:${e.stack}`, this.#createConsumerBindingMessageMetadata());
     }
   }
 
@@ -782,6 +802,7 @@ class Consumer {
     }
 
     this.#fetchInProgress = true;
+    this.#logger.debug(`Attempting to fetch ${this.#messageCache.maxSize} messages to the message cache`, this.#createConsumerBindingMessageMetadata());
     return new Promise((resolve, reject) => {
       this.#internalClient.consume(this.#messageCache.maxSize, (err, messages) => {
         this.#fetchInProgress = false;
@@ -828,6 +849,7 @@ class Consumer {
     }
 
     this.#fetchInProgress = true;
+    this.#logger.debug(`Attempting to fetch ${this.#messageCache.maxSize} messages to the message cache`, this.#createConsumerBindingMessageMetadata());
     return new Promise((resolve, reject) => {
       this.#internalClient.consume(this.#messageCache.maxSize, (err, messages) => {
         this.#fetchInProgress = false;
@@ -968,6 +990,7 @@ class Consumer {
     });
 
     this.#storedSubscriptions = subscription.replace ? topics : this.#storedSubscriptions.concat(topics);
+    this.#logger.debug(`${subscription.replace ? 'Replacing' : 'Adding'} topics [${topics.join(', ')}] to subscription`, this.#createConsumerBindingMessageMetadata());
     this.#internalClient.subscribe(this.#storedSubscriptions);
   }
 
@@ -1043,12 +1066,12 @@ class Consumer {
        * However, we don't do this inside the catch, but just outside it. This is because throwing an
        * error is not the only case where we might want to seek back.
        *
-       * So - do nothing but a debug log, but at this point eachMessageProcessed is false.
+       * So - do nothing but a log, but at this point eachMessageProcessed is false.
+       * TODO: log error only if error type is not KafkaJSError and if no pause() has been called, else log debug.
        */
-      this.#logger.debug(`Consumer encountered error while processing message. Error details: ${e}: ${e.stack}. The same message may be reprocessed.`);
-
-      /* TODO: log error if error type is not KafkaJSError and if no pause() has been called */
-      this.#logger.error(`Consumer encountered error while processing message. Error details: ${e}: ${e.stack}. The same message may be reprocessed.`);
+      this.#logger.error(
+        `Consumer encountered error while processing message. Error details: ${e}: ${e.stack}. The same message may be reprocessed.`,
+        this.#createConsumerBindingMessageMetadata());
     }
 
     /* If the message is unprocessed, due to an error, or because the user has not resolved it, we seek back. */
@@ -1067,8 +1090,7 @@ class Consumer {
         this.#lastConsumedOffsets.set(partitionKey(m), Number(m.offset) + 1);
       } catch (e) {
         /* Not much we can do, except log the error. */
-        if (this.#logger)
-          this.#logger.error(`Consumer encountered error while storing offset. Error details: ${JSON.stringify(e)}`);
+        this.#logger.error(`Consumer encountered error while storing offset. Error details: ${JSON.stringify(e)}`, this.#createConsumerBindingMessageMetadata());
       }
     }
 
@@ -1123,13 +1145,13 @@ class Consumer {
        * if the user has not called `resolveOffset` manually in case of using eachBatch without
        * eachBatchAutoResolve being set.
        *
-       * So - do nothing but a debug log, but at this point eachMessageProcessed needs to be false unless
+       * So - do nothing but a log, but at this point eachMessageProcessed needs to be false unless
        * the user has explicitly marked it as true.
+       * TODO: log error only if error type is not KafkaJSError and if no pause() has been called, else log debug.
        */
-      this.#logger.debug(`Consumer encountered error while processing message. Error details: ${e}: ${e.stack}. The same message may be reprocessed.`);
-
-      /* TODO: log error if error type is not KafkaJSError and if no pause() has been called */
-      this.#logger.error(`Consumer encountered error while processing message. Error details: ${e}: ${e.stack}. The same message may be reprocessed.`);
+      this.#logger.error(
+        `Consumer encountered error while processing message. Error details: ${e}: ${e.stack}. The same message may be reprocessed.`,
+        this.#createConsumerBindingMessageMetadata());
 
       /* The value of eachBatchAutoResolve is not important. The only place where a message is marked processed
        * despite an error is if the user says so, and the user can use resolveOffset for both the possible
@@ -1177,6 +1199,7 @@ class Consumer {
       /* Invalidate the message cache if needed */
       const locallyStale = this.#messageCache.popLocallyStale();
       if (this.#messageCache.isStale()) {
+        this.#logger.debug("Scheduling worker termination", this.#createConsumerBindingMessageMetadata());
         this.#workerTerminationScheduled = true;
         break;
       } else if (locallyStale.length !== 0) {
@@ -1191,7 +1214,7 @@ class Consumer {
         /* Since this error cannot be exposed to the user in the current situation, just log and retry.
          * This is due to restartOnFailure being set to always true. */
         if (this.#logger)
-          this.#logger.error(`Consumer encountered error while consuming. Retrying. Error details: ${e} : ${e.stack}`);
+          this.#logger.error(`Consumer encountered error while consuming. Retrying. Error details: ${e} : ${e.stack}`, this.#createConsumerBindingMessageMetadata());
       });
 
       nextIdx = -1;
@@ -1229,14 +1252,15 @@ class Consumer {
     while (!this.#disconnectStarted) {
       this.#workerTerminationScheduled = false;
       const workersToSpawn = Math.max(1, Math.min(this.#concurrency, this.#partitionCount));
+      this.#logger.debug(`Spawning ${workersToSpawn} workers`, this.#createConsumerBindingMessageMetadata());
       this.#workers =
         Array(workersToSpawn)
           .fill()
           .map((_, i) =>
             this.#worker(config, perMessageProcessor.bind(this), fetcher.bind(this))
               .catch(e => {
                 if (this.#logger)
-                  this.#logger.error(`Worker ${i} encountered an error: ${e}:${e.stack}`);
+                  this.#logger.error(`Worker ${i} encountered an error: ${e}:${e.stack}`, this.#createConsumerBindingMessageMetadata());
               }));
 
       /* Best we can do is log errors on worker issues - handled by the catch block above. */
@@ -1427,7 +1451,10 @@ class Consumer {
       /* It's assumed that topicPartition is already assigned, and thus can be seeked to and committed to.
        * Errors are logged to detect bugs in the internal code. */
       /* TODO: is it worth awaiting seeks to finish? */
-      this.#internalClient.seek(topicPartitionOffset, 0, err => err ? this.#logger.error(err) : null);
+      this.#internalClient.seek(topicPartitionOffset, 0, err => {
+        if (err)
+          this.#logger.error(`Error while calling seek from within seekInternal: ${err}`, this.#createConsumerBindingMessageMetadata());
+      });
       offsetsToCommit.push({
         topic: topicPartition.topic,
         partition: topicPartition.partition,
@@ -1539,6 +1566,8 @@ class Consumer {
       throw new error.KafkaJSError('Pause can only be called while connected.', { code: error.ErrorCodes.ERR__STATE });
     }
 
+    this.#logger.debug(`Pausing ${topics.length} topics`, this.#createConsumerBindingMessageMetadata());
+
     const toppars = [];
     for (let topic of topics) {
       if (typeof topic.topic !== 'string') {
@@ -1612,6 +1641,8 @@ class Consumer {
       throw new error.KafkaJSError('Resume can only be called while connected.', { code: error.ErrorCodes.ERR__STATE });
     }
 
+    this.#logger.debug(`Resuming ${topics.length} topics`, this.#createConsumerBindingMessageMetadata());
+
     const toppars = [];
     for (let topic of topics) {
       if (typeof topic.topic !== 'string') {
@@ -1677,6 +1708,8 @@ class Consumer {
 
     this.#disconnectStarted = true;
     this.#workerTerminationScheduled = true;
+
+    this.#logger.debug("Signalling disconnection attempt to workers", this.#createConsumerBindingMessageMetadata());
     while (!(await acquireOrLog(this.#lock, this.#logger))); /* Just retry... */
 
     this.#state = ConsumerState.DISCONNECTING;
@@ -1691,6 +1724,7 @@ class Consumer {
           return;
         }
         this.#state = ConsumerState.DISCONNECTED;
+        this.#logger.info("Consumer disconnected", this.#createConsumerBindingMessageMetadata());
         resolve();
       };
       this.#internalClient.disconnect(cb);