Fix log level config in light of binding logs

* Also add rule for no trailing spaces and enforce it
* Fix log level config in light of binding logs

Author: milindl

eslint.config.js

@@ -21,6 +21,7 @@ const ckjsSpecificSettings = {
         "no-caller": "error",
         "no-new": "error",
         "no-eq-null": "error",
+        "no-trailing-spaces": "error",
         "no-constant-condition": "off",
         "semi": "error"
     }

lib/kafka-consumer.js

@@ -345,7 +345,7 @@ KafkaConsumer.prototype.assignments = function() {
  *
  * @note This method should only be called from within the rebalance callback
  * when partitions are revoked.
- * 
+ *
  * @return {boolean} true if assignment was lost
  */
 

lib/kafkajs/_admin.js

@@ -116,9 +116,23 @@ class Admin {
   #finalizedConfig() {
     let compatibleConfig = this.#kafkaJSToAdminConfig(this.#userConfig.kafkaJS);
 
-    /* Set the logger's level in case we're not in compatibility mode - just set it to DEBUG, the broadest
-     * log level, as librdkafka will control the granularity. */
-    if (!compatibleConfig || Object.keys(compatibleConfig).length === 0) {
+    /* There can be multiple different and conflicting config directives for setting the log level:
+     * 1. If there's a kafkaJS block:
+     *   a. If there's a logLevel directive in the kafkaJS block, set the logger level accordingly.
+     *   b. If there's no logLevel directive, set the logger level to INFO.
+     *   (both these are already handled in the conversion method above).
+     * 2. If there is a log_level or debug directive in the main config, set the logger level accordingly.
+     *    !This overrides any different value provided in the kafkaJS block!
+     *   a. If there's a log_level directive, set the logger level accordingly.
+     *   b. If there's a debug directive, set the logger level to DEBUG regardless of anything else. This is because
+     *      librdkafka ignores log_level if debug is set, and our behaviour should be identical.
+     * 3. There's nothing at all. Take no action in this case, let the logger use its default log level.
+     */
+    if (Object.hasOwn(this.#userConfig, 'log_level')) {
+      this.#logger.setLogLevel(severityToLogLevel[this.#userConfig.log_level]);
+    }
+
+    if (Object.hasOwn(this.#userConfig, 'debug')) {
       this.#logger.setLogLevel(logLevel.DEBUG);
     }
 

lib/kafkajs/_common.js

@@ -443,7 +443,6 @@ function kafkaJSToRdKafkaConfig(config) {
         rdkafkaConfig["log_level"] = 6 /* LOG_INFO */;
         break;
       case logLevel.DEBUG:
-        rdkafkaConfig["debug"] = "all" /* Turn on debug logs for everything, otherwise this log level is not useful*/;
         rdkafkaConfig["log_level"] = 7 /* LOG_DEBUG */;
         break;
       default:
@@ -639,7 +638,7 @@ class DeferredPromise extends Promise{
    * that takes the same parameter a normal Promise constructor does.
    * The DeferredPromise cannot be rejected to avoid unhandled rejections
    * entirely.
-   * @param {(resolve: (value: any) => void, reject: (error: Error) => void) => void} resolver 
+   * @param {(resolve: (value: any) => void, reject: (error: Error) => void) => void} resolver
    */
   constructor(resolver) {
     let resolveF;
@@ -662,12 +661,12 @@ class DeferredPromise extends Promise{
 /**
  * Utility class for time related functions
  */
-class Timer { 
+class Timer {
   /**
    * Function that resolves when the given timeout is reached
    * or the passed promise resolves, when it's passed, clearing the timeout
    * in any case.
-   * 
+   *
    * @param {number} timeoutMs The timeout in milliseconds.
    * @param {Promise|undefined} promise The promise to wait for,
    * alternatively to the timeout, or `undefined` to just wait for the timeout.
@@ -698,13 +697,13 @@ class Timer {
 class Lock {
   // Total number of readers, not increases when already holding a write lock
   #readers = 0;
-  
+
   // Total number of writers, increased only by a single write and
   // its reentrant calls
   #writers = 0;
 
   #asyncLocalStorage = new AsyncLocalStorage();
-  
+
   // Promise to resolve and recreate when there are no readers or writers
   // This is used to notify all waiting writers so at least one can proceed.
   // It's also used to notify all waiting readers so they can can check
@@ -789,7 +788,7 @@ class Lock {
     this.#notifyZeroReadersAndWriters();
   }
 
-  /** 
+  /**
    *  Acquire a write (exclusive) lock while executing
    *  the given task.
    *  @param {function} task The task to execute.
@@ -807,8 +806,8 @@ class Lock {
     await this.#runAsyncStack(1, withWriteLock);
   }
 
-  
-  /** 
+
+  /**
    *  Acquire a read (shared) lock while executing
    *  the given task.
    *  @param {function} task The task to execute.

lib/kafkajs/_consumer.js

@@ -336,7 +336,7 @@ class Consumer {
       else
         this.#addPendingOperation(() => this.#unassign(userAssignment));
     };
-     
+
     try {
       err = LibrdKafkaError.create(err);
       const userSpecifiedRebalanceCb = this.#userConfig['rebalance_cb'];
@@ -549,13 +549,22 @@ class Consumer {
     /* Creates an rdkafka config based off the kafkaJS block. Switches to compatibility mode if the block exists. */
     let compatibleConfig = this.#kafkaJSToConsumerConfig(this.#userConfig.kafkaJS);
 
-    /* Set the logger's level in case we're not in compatibility mode - just set it to DEBUG, the broadest
-     * log level, as librdkafka will control the granularity. */
-    if (!compatibleConfig || Object.keys(compatibleConfig).length === 0) {
-      this.#logger.setLogLevel(logLevel.DEBUG);
+    /* There can be multiple different and conflicting config directives for setting the log level:
+     * 1. If there's a kafkaJS block:
+     *   a. If there's a logLevel directive in the kafkaJS block, set the logger level accordingly.
+     *   b. If there's no logLevel directive, set the logger level to INFO.
+     *   (both these are already handled in the conversion method above).
+     * 2. If there is a log_level or debug directive in the main config, set the logger level accordingly.
+     *    !This overrides any different value provided in the kafkaJS block!
+     *   a. If there's a log_level directive, set the logger level accordingly.
+     *   b. If there's a debug directive, set the logger level to DEBUG regardless of anything else. This is because
+     *      librdkafka ignores log_level if debug is set, and our behaviour should be identical.
+     * 3. There's nothing at all. Take no action in this case, let the logger use its default log level.
+     */
+    if (Object.hasOwn(this.#userConfig, 'log_level')) {
+      this.#logger.setLogLevel(severityToLogLevel[this.#userConfig.log_level]);
     }
 
-    /* Even if we are in compability mode, setting a 'debug' in the main config must override the logger's level. */
     if (Object.hasOwn(this.#userConfig, 'debug')) {
       this.#logger.setLogLevel(logLevel.DEBUG);
     }
@@ -568,8 +577,8 @@ class Consumer {
     /* Certain properties that the user has set are overridden. We use trampolines to accommodate the user's callbacks.
      * TODO: add trampoline method for offset commit callback. */
     rdKafkaConfig['offset_commit_cb'] = true;
-    rdKafkaConfig['rebalance_cb'] = (err, assignment) => this.#rebalanceCallback(err, assignment).catch(e => 
-      { 
+    rdKafkaConfig['rebalance_cb'] = (err, assignment) => this.#rebalanceCallback(err, assignment).catch(e =>
+      {
         if (this.#logger)
           this.#logger.error(`Error from rebalance callback: ${e.stack}`);
       });
@@ -1307,7 +1316,7 @@ class Consumer {
   }
 
   async #checkMaxPollIntervalNotExceeded(now) {
-    const maxPollExpiration = this.#lastFetchClockNs + 
+    const maxPollExpiration = this.#lastFetchClockNs +
       BigInt((this.#cacheExpirationTimeoutMs + this.#maxPollIntervalMs)
               * 1e6);
 
@@ -1317,15 +1326,15 @@ class Consumer {
     await Timer.withTimeout(interval,
       this.#maxPollIntervalRestart);
     now = hrtime.bigint();
-    
+
     if (now > (maxPollExpiration - 1000000n)) {
       this.#markBatchPayloadsStale(this.assignment());
     }
   }
 
   /**
    * Clears the cache and resets the positions when
-   * the internal client hasn't been polled for more than 
+   * the internal client hasn't been polled for more than
    * max poll interval since the last fetch.
    * After that it waits until barrier is reached or
    * max poll interval is reached. In the latter case it
@@ -1334,11 +1343,11 @@ class Consumer {
   async #cacheExpirationLoop() {
     while (!this.#workerTerminationScheduled.resolved) {
       let now = hrtime.bigint();
-      const cacheExpiration = this.#lastFetchClockNs + 
+      const cacheExpiration = this.#lastFetchClockNs +
         BigInt(this.#cacheExpirationTimeoutMs * 1e6);
 
       if (now > cacheExpiration) {
-        this.#addPendingOperation(() => 
+        this.#addPendingOperation(() =>
           this.#clearCacheAndResetPositions());
         await this.#checkMaxPollIntervalNotExceeded(now);
         break;
@@ -1558,7 +1567,7 @@ class Consumer {
 
     // Uncomment to test an additional delay in seek
     // await Timer.withTimeout(1000);
-    
+
     const seekedPartitions = [];
     const pendingSeeks = new Map();
     const assignmentSet = new Set();
@@ -1608,7 +1617,7 @@ class Consumer {
     /* Offsets are committed on seek only when in compatibility mode. */
     if (offsetsToCommit.length !== 0 && this.#internalConfig['enable.auto.commit']) {
       await this.#commitOffsetsUntilNoStateErr(offsetsToCommit);
-    }      
+    }
   }
 
   #markBatchPayloadsStale(topicPartitions) {
@@ -1828,7 +1837,7 @@ class Consumer {
     }
     flattenedToppars.map(partitionKey).
       forEach(key => this.#pausedPartitions.delete(key));
-    
+
     this.#addPendingOperation(() =>
       this.#resumeInternal(flattenedToppars));
   }

lib/kafkajs/_consumer_cache.js

@@ -61,7 +61,7 @@ class PerPartitionMessageCache {
 
 /**
  * MessageCache defines a dynamically sized cache for messages.
- * Internally, it uses PerPartitionMessageCache to store messages for each partition. 
+ * Internally, it uses PerPartitionMessageCache to store messages for each partition.
  */
 class MessageCache {
     #size;
@@ -90,7 +90,7 @@ class MessageCache {
 
     /**
      * Assign a new partition to the consumer, if available.
-     * 
+     *
      * @returns {PerPartitionMessageCache} - the partition assigned to the consumer, or null if none available.
      */
     #assignNewPartition() {
@@ -105,7 +105,7 @@ class MessageCache {
 
     /**
      * Remove an empty partition from the cache.
-     * 
+     *
      * @param {PerPartitionMessageCache} ppc The partition to remove from the cache.
      */
     #removeEmptyPartition(ppc) {
@@ -118,7 +118,7 @@ class MessageCache {
     /**
      * Add a single message to a PPC.
      * In case the PPC does not exist, it is created.
-     * 
+     *
      * @param {Object} message - the message to add to the cache.
      */
     #add(message) {
@@ -172,7 +172,7 @@ class MessageCache {
     /**
      * Adds many messages into the cache, partitioning them as per their toppar.
      * Increases cache size by the number of messages added.
-     * 
+     *
      * @param {Array} messages - the messages to add to the cache.
      */
     addMessages(messages) {
@@ -183,9 +183,9 @@ class MessageCache {
 
     /**
      * Allows returning the PPC without asking for another message.
-     * 
+     *
      * @param {PerPartitionMessageCache} ppc - the partition to return.
-     * 
+     *
      * @note this is a no-op if the PPC is not assigned.
      */
     return(ppc) {
@@ -203,7 +203,7 @@ class MessageCache {
      *
      * If the current PPC is exhausted, it moves to the next PPC.
      * If all PPCs are exhausted, it returns null.
-     * 
+     *
      * @param {PerPartitionMessageCache} ppc - after a consumer has consumed a message, it must return the PPC back to us via this parameter.
      *                       otherwise, no messages from that topic partition will be consumed.
      * @returns {Array} - the next message in the cache, or null if none exists, and the corresponding PPC.
@@ -245,7 +245,7 @@ class MessageCache {
         if (!nextN.length)
             return this.nextN(null, size);
 
-        this.#size -= nextN.length;	
+        this.#size -= nextN.length;
         return [nextN, ppc];
     }
 

lib/kafkajs/_linked-list.js

@@ -62,8 +62,8 @@ class LinkedList {
     /**
      * Removes given node from the list,
      * if it is not already removed.
-     * 
-     * @param {LinkedListNode} node 
+     *
+     * @param {LinkedListNode} node
      */
     remove(node) {
         if (node._removed) {
@@ -89,7 +89,7 @@ class LinkedList {
     /**
      * Removes the first node from the list and returns it,
      * or null if the list is empty.
-     * 
+     *
      * @returns {any} The value of the first node in the list or null.
      */
     removeFirst() {
@@ -105,7 +105,7 @@ class LinkedList {
     /**
      * Removes the last node from the list and returns its value,
      * or null if the list is empty.
-     * 
+     *
      * @returns {any} The value of the last node in the list or null.
      */
     removeLast() {
@@ -120,31 +120,31 @@ class LinkedList {
 
     /**
      * Add a new node to the beginning of the list and returns it.
-     * 
-     * @param {any} value 
+     *
+     * @param {any} value
      * @returns {LinkedListNode} The new node.
      */
-    addFirst(value) {    
+    addFirst(value) {
         const node = new LinkedListNode(value);
         return this.#insertInBetween(node, null,
             this._head);
     }
 
     /**
      * Add a new node to the end of the list and returns it.
-     * 
+     *
      * @param {any} value Node value.
      * @returns {LinkedListNode} The new node.
      */
-    addLast(value) {    
+    addLast(value) {
         const node = new LinkedListNode(value);
         return this.#insertInBetween(node, this._tail, null);
     }
 
     /**
      * Add a new node before the given node and returns it.
      * Given node must not be removed.
-     * 
+     *
      * @param {LinkedListNode} node Reference node.
      * @param {any} value New node value.
      * @returns {LinkedListNode} The new node.
@@ -159,7 +159,7 @@ class LinkedList {
     /**
      * Add a new node after the given node and returns it.
      * Given node must not be removed.
-     * 
+     *
      * @param {LinkedListNode} node Reference node.
      * @param {any} value New node value.
      * @returns {LinkedListNode} The new node.
@@ -173,7 +173,7 @@ class LinkedList {
 
     /**
      * Concatenates the given list to the end of this list.
-     * 
+     *
      * @param {LinkedList} list List to concatenate.
      */
     concat(list) {