[fix] pass string as first arg to user logger on error events

lib/kafkajs/_admin.js

@@ -204,9 +204,8 @@ class Admin {
     if (this.#state < AdminState.CONNECTED) {
       if (!this.#connectionError)
         this.#connectionError = err;
-    } else {
-      this.#logger.error(`Error: ${err.message}`, this.#createAdminBindingMessageMetadata());
     }
+    this.#logger.error(`Error: ${err.message}`, this.#createAdminBindingMessageMetadata());
   }
 
   /**

lib/kafkajs/_common.js

@@ -130,7 +130,7 @@ class DefaultLogger {
  */
 function createBindingMessageMetadata(clientName) {
   return {
-    name: clientName,
+    name: clientName || '',
     fac: 'BINDING',
     timestamp: Date.now(),
   };

lib/kafkajs/_consumer.js

@@ -774,9 +774,8 @@ class Consumer {
     if (this.#state < ConsumerState.CONNECTED) {
       if (!this.#connectionError)
         this.#connectionError = err;
-    } else {
-      this.#logger.error(err, this.#createConsumerBindingMessageMetadata());
     }
+    this.#logger.error(`Error: ${err.message}`, this.#createConsumerBindingMessageMetadata());
   }
 
 

lib/kafkajs/_producer.js

@@ -376,7 +376,7 @@ class Producer {
       if (!this.#connectionError)
         this.#connectionError = err;
     }
-    this.#logger.error(err, this.#createProducerBindingMessageMetadata());
+    this.#logger.error(`Error: ${err.message}`, this.#createProducerBindingMessageMetadata());
   }
 
   /**

test/promisified/unit/logger.spec.js

@@ -0,0 +1,121 @@
+jest.mock('../../../lib/rdkafka', () => {
+  const { EventEmitter } = require('events');
+
+  function makeStubClient(name) {
+    const client = new EventEmitter();
+    client.name = name;
+    client.connect = jest.fn();
+    client.disconnect = jest.fn();
+    client.setPollInBackground = jest.fn();
+    client.setDefaultIsTimeoutOnlyForFirstMessage = jest.fn();
+    client.setDefaultConsumeTimeout = jest.fn();
+    return client;
+  }
+
+  const captured = { producer: null, consumer: null, admin: null };
+
+  function ProducerCtor() {
+    captured.producer = makeStubClient('rdkafka#producer-test');
+    return captured.producer;
+  }
+
+  function KafkaConsumerCtor() {
+    captured.consumer = makeStubClient('rdkafka#consumer-test');
+    return captured.consumer;
+  }
+
+  return {
+    Producer: ProducerCtor,
+    KafkaConsumer: KafkaConsumerCtor,
+    AdminClient: {
+      create: jest.fn((_config, listeners) => {
+        const stub = {
+          name: 'rdkafka#admin-test',
+          on: jest.fn(),
+          disconnect: jest.fn(),
+          _listeners: listeners,
+        };
+        captured.admin = stub;
+        if (listeners && typeof listeners.ready === 'function') {
+          listeners.ready();
+        }
+        return stub;
+      }),
+    },
+    CODES: { ERRORS: {} },
+    __captured: captured,
+  };
+});
+
+const { Kafka } = require('../../../lib/kafkajs');
+const LibrdKafkaError = require('../../../lib/error');
+const RdKafkaMock = require('../../../lib/rdkafka');
+
+function makeLogger() {
+  const logger = {
+    info: jest.fn(),
+    warn: jest.fn(),
+    error: jest.fn(),
+    debug: jest.fn(),
+    setLogLevel: jest.fn(),
+  };
+  logger.namespace = jest.fn(() => logger);
+  return logger;
+}
+
+function brokerError() {
+  return LibrdKafkaError.create(new Error('Local: broker transport failure'));
+}
+
+function expectStringFirstArg(mockFn, err) {
+  expect(mockFn).toHaveBeenCalled();
+  const [firstArg, secondArg] = mockFn.mock.calls[mockFn.mock.calls.length - 1];
+  expect(firstArg).toEqual(`Error: ${err.message}`);
+  expect(secondArg).toEqual(expect.objectContaining({
+    fac: 'BINDING',
+    name: expect.any(String),
+    timestamp: expect.any(Number),
+  }));
+}
+
+describe('user-supplied logger receives a string as the first arg on error callbacks', () => {
+  beforeEach(() => {
+    RdKafkaMock.__captured.producer = null;
+    RdKafkaMock.__captured.consumer = null;
+    RdKafkaMock.__captured.admin = null;
+  });
+
+  it('Producer #errorCb passes a string to logger.error', async () => {
+    const logger = makeLogger();
+    const producer = new Kafka({ kafkaJS: { brokers: ['x:1'], logger } }).producer();
+    producer.connect().catch(() => {});
+
+    const err = brokerError();
+    RdKafkaMock.__captured.producer.emit('event.error', err);
+
+    expectStringFirstArg(logger.error, err);
+  });
+
+  it('Consumer #errorCb passes a string to logger.error', async () => {
+    const logger = makeLogger();
+    const kafka = new Kafka({ kafkaJS: { brokers: ['x:1'], logger } });
+    const consumer = kafka.consumer({ kafkaJS: { groupId: 'g' } });
+    consumer.connect().catch(() => {});
+
+    const err = brokerError();
+    RdKafkaMock.__captured.consumer.emit('event.error', err);
+
+    expectStringFirstArg(logger.error, err);
+  });
+
+  it('Admin #errorCb passes a string to logger.error', async () => {
+    const logger = makeLogger();
+    const admin = new Kafka({ kafkaJS: { brokers: ['x:1'], logger } }).admin();
+    await admin.connect();
+
+    const err = brokerError();
+    RdKafkaMock.__captured.admin._listeners.error(err);
+
+    expectStringFirstArg(logger.error, err);
+  });
+});