Fix consumeMessages test and add TEST_DEBUG env var

Author: milindl

lib/kafkajs/_consumer.js

@@ -322,6 +322,11 @@ class Consumer {
       this.#logger.setLogLevel(logLevel.DEBUG);
     }
 
+    /* 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);
+    }
+
     let rdKafkaConfig = Object.assign(compatibleConfig, this.#userConfig);
 
     /* Delete properties which are already processed, or cannot be passed to node-rdkafka */

test/promisified/consumer/consumeMessages.spec.js

@@ -642,7 +642,6 @@ describe('Consumer', () => {
             async () => {
                 // Seed the topic with some messages. We don't need a tx producer for this.
                 await producer.connect();
-
                 const partition = 0;
                 const messages = generateMessages().map(message => ({
                     ...message,
@@ -691,7 +690,7 @@ describe('Consumer', () => {
                     eachMessage,
                 })
 
-                // Consume pre-produced messages.
+                // 2. Consume pre-produced messages.
 
                 const number = messages.length;
                 await waitForMessages(messagesConsumed, {
@@ -729,8 +728,8 @@ describe('Consumer', () => {
                 await consumer.connect();
                 await consumer.subscribe({ topic: topicName });
 
-                messagesConsumed = []
-                uncommittedOffsetsPerMessage = []
+                messagesConsumed = [];
+                uncommittedOffsetsPerMessage = [];
 
                 consumer.run({ eachMessage })
 
@@ -802,19 +801,6 @@ describe('Consumer', () => {
                 // Consume produced messages.
                 await waitForMessages(messagesConsumed, { number: messages.length });
 
-                // Restart consumer - we cannot stop it, so we recreate it.
-                await consumer.disconnect();
-
-                consumer = createConsumer({
-                    groupId,
-                    maxWaitTimeInMs: 100,
-                    fromBeginning: true,
-                    autoCommit: false,
-                });
-
-                await consumer.connect();
-                await consumer.subscribe({ topic: topicName });
-
                 expect(messagesConsumed[0].value.toString()).toMatch(/value-0/);
                 expect(messagesConsumed[99].value.toString()).toMatch(/value-99/);
                 expect(uncommittedOffsetsPerMessage).toHaveLength(messagesConsumed.length);
@@ -833,13 +819,25 @@ describe('Consumer', () => {
                 });
                 await txnToAbort.abort()
 
-                // Restart consumer
+                /* Restart consumer - we cannot stop it, so we recreate it. */
                 messagesConsumed = []
                 uncommittedOffsetsPerMessage = []
 
+                await consumer.disconnect();
+
+                consumer = createConsumer({
+                    groupId,
+                    maxWaitTimeInMs: 100,
+                    fromBeginning: true,
+                    autoCommit: false,
+                });
+
+                await consumer.connect();
+                await consumer.subscribe({ topic: topicName });
+
                 consumer.run({
-                    eachMessage
-                })
+                    eachMessage,
+                });
 
                 await waitForMessages(messagesConsumed, { number: 1 });
                 expect(messagesConsumed[0].value.toString()).toMatch(/value-0/)

test/promisified/testhelpers.js

@@ -1,24 +1,37 @@
 const crypto = require('crypto');
 const process = require('process');
+const { logLevel } = require('../../lib/kafkajs');
 const { Kafka } = require('../../lib').KafkaJS;
 
 // TODO: pick this up from a file
 const clusterInformation = {
     brokers: process.env.KAFKA_HOST ? process.env.KAFKA_HOST.split(',') : ['localhost:9092'],
 };
 
+const debug = process.env.TEST_DEBUG;
+
+function makeConfig(config) {
+    const kafkaJs =  Object.assign(config, clusterInformation);
+    const common = {};
+    if (debug) {
+        common['debug'] = debug;
+    }
+
+    return Object.assign(common, { kafkaJs });
+}
+
 function createConsumer(config) {
-    const kafka = new Kafka({ kafkaJs: Object.assign(config, clusterInformation) });
+    const kafka = new Kafka(makeConfig(config));
     return kafka.consumer();
 }
 
 function createProducer(config) {
-    const kafka = new Kafka({ kafkaJs: Object.assign(config, clusterInformation) });
+    const kafka = new Kafka(makeConfig(config));
     return kafka.producer();
 }
 
 function createAdmin(config) {
-    const kafka = new Kafka({ kafkaJs: Object.assign(config, clusterInformation) });
+    const kafka = new Kafka(makeConfig(config));
     return kafka.admin();
 }