Add functional changes for kafkaJs block

Author: milindl

lib/kafkajs/_common.js

@@ -97,11 +97,11 @@ function createReplacementErrorMessage(cOrP, fnCall, property, propertyVal, repl
   }
   return `'${property}' is not supported as a property to '${fnCall}', but must be passed to the ${cOrP} during creation.\n` +
     `Before: \n` +
-    `\tconst ${cOrP} = kafka.${cOrP}({ ... });\n` +
+    `\tconst ${cOrP} = kafka.${cOrP}({ kafkaJs: { ... }, });\n` +
     `\tawait ${cOrP}.connect();\n` +
     `\t${cOrP}.${fnCall}({ ${propertyVal}, ... });\n` +
     `After: \n` +
-    `\tconst ${cOrP} = kafka.${cOrP}({ ${replacementVal}, ... });\n` +
+    `\tconst ${cOrP} = kafka.${cOrP}({ kafkaJs: { ${replacementVal}, ... }, });\n` +
     `\tawait ${cOrP}.connect();\n` +
     `\t${cOrP}.${fnCall}({ ... });\n` +
     (isLK ? `For more details on what can be used inside the rdKafka block, see https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md\n` : '');
@@ -179,24 +179,28 @@ const CompatibilityErrorMessages = Object.freeze({
 /**
  * Converts the common configuration from KafkaJS to a format that can be used by node-rdkafka.
  * @param {object} config
- * @returns {{globalConfig: import("../../types/config").ConsumerGlobalConfig|import("../../types/config").ProducerTopicConfig, topicConfig: import("../../types/config").ConsumerTopicConfig|import("../../types/config").ProducerTopicConfig}}
+ * @returns {import('../../types/config').ProducerGlobalConfig | import('../../types/config').ConsumerGlobalConfig} the converted configuration
  * @throws {error.KafkaJSError} if the configuration is invalid.
  *                              The error code will be ERR__INVALID_ARG in case of invalid arguments or features that are not supported.
  *                              The error code will be ERR__NOT_IMPLEMENTED in case of features that are not yet implemented.
  */
-async function kafkaJSToRdKafkaConfig(config) {
-  const globalConfig = {};
-  const topicConfig = {};
-
-  if (!Array.isArray(config["brokers"])) {
-    throw new error.KafkaJSError(CompatibilityErrorMessages.brokerString(), {
-      code: error.ErrorCodes.ERR__INVALID_ARG,
-    });
+function kafkaJSToRdKafkaConfig(config) {
+  /* Since the kafkaJs block is specified, we operate in
+   * kafkaJs compatibility mode. That means we change the defaults
+   * match the kafkaJs defaults. */
+  const rdkafkaConfig = {};
+
+  if (Object.hasOwn(config, "brokers")) {
+    if (!Array.isArray(config["brokers"])) {
+      throw new error.KafkaJSError(CompatibilityErrorMessages.brokerString(), {
+        code: error.ErrorCodes.ERR__INVALID_ARG,
+      });
+    }
+    rdkafkaConfig["bootstrap.servers"] = config["brokers"].join(",");
   }
-  globalConfig["bootstrap.servers"] = config["brokers"].join(",");
 
   if (Object.hasOwn(config, "clientId")) {
-    globalConfig["client.id"] = config.clientId;
+    rdkafkaConfig["client.id"] = config.clientId;
   }
 
   let withSASL = false;
@@ -224,18 +228,18 @@ async function kafkaJSToRdKafkaConfig(config) {
       });
     }
 
-    globalConfig["sasl.mechanism"] = mechanism;
-    globalConfig["sasl.username"] = sasl.username;
-    globalConfig["sasl.password"] = sasl.password;
+    rdkafkaConfig["sasl.mechanism"] = mechanism;
+    rdkafkaConfig["sasl.username"] = sasl.username;
+    rdkafkaConfig["sasl.password"] = sasl.password;
     withSASL = true;
   }
 
   if (Object.hasOwn(config, "ssl") && config.ssl && withSASL) {
-    globalConfig["security.protocol"] = "sasl_ssl";
+    rdkafkaConfig["security.protocol"] = "sasl_ssl";
   } else if (withSASL) {
-    globalConfig["security.protocol"] = "sasl_plaintext";
+    rdkafkaConfig["security.protocol"] = "sasl_plaintext";
   } else if (Object.hasOwn(config, "ssl") && config.ssl) {
-    globalConfig["security.protocol"] = "ssl";
+    rdkafkaConfig["security.protocol"] = "ssl";
   }
 
   /* TODO: add best-effort support for ssl besides just true/false */
@@ -246,14 +250,14 @@ async function kafkaJSToRdKafkaConfig(config) {
   }
 
   if (Object.hasOwn(config, "requestTimeout")) {
-    globalConfig["socket.timeout.ms"] = config.requestTimeout;
+    rdkafkaConfig["socket.timeout.ms"] = config.requestTimeout;
   } else {
     /* KafkaJS default */
-    globalConfig["socket.timeout.ms"] = 30000;
+    rdkafkaConfig["socket.timeout.ms"] = 30000;
   }
 
   if (Object.hasOwn(config, "enforceRequestTimeout") && !config.enforceRequestTimeout) {
-    globalConfig["socket.timeout.ms"] = 300000;
+    rdkafkaConfig["socket.timeout.ms"] = 300000;
   }
 
   const connectionTimeout = config.connectionTimeout ?? 1000;
@@ -262,14 +266,14 @@ async function kafkaJSToRdKafkaConfig(config) {
 
   /* The minimum value for socket.connection.setup.timeout.ms is 1000. */
   totalConnectionTimeout = Math.max(totalConnectionTimeout, 1000);
-  globalConfig["socket.connection.setup.timeout.ms"] = totalConnectionTimeout;
+  rdkafkaConfig["socket.connection.setup.timeout.ms"] = totalConnectionTimeout;
 
   const retry = config.retry ?? {};
   const { maxRetryTime, initialRetryTime, factor, multiplier, retries, restartOnFailure } = retry;
 
-  globalConfig["retry.backoff.max.ms"] = maxRetryTime ?? 30000;
-  globalConfig["retry.backoff.ms"] = initialRetryTime ?? 300;
-  globalConfig["retries"] = retries ?? 5;
+  rdkafkaConfig["retry.backoff.max.ms"] = maxRetryTime ?? 30000;
+  rdkafkaConfig["retry.backoff.ms"] = initialRetryTime ?? 300;
+  rdkafkaConfig["retries"] = retries ?? 5;
 
   if ((typeof factor === 'number') || (typeof multiplier === 'number')) {
     throw new error.KafkaJSError(CompatibilityErrorMessages.retryFactorMultiplier(), {
@@ -295,6 +299,7 @@ async function kafkaJSToRdKafkaConfig(config) {
     });
   }
 
+  rdkafkaConfig["log_level"] = 6 /* LOG_INFO - default in KafkaJS compatibility mode. */;
   if (Object.hasOwn(config, "logLevel")) {
     let setLevel = config.logLevel;
 
@@ -303,19 +308,20 @@ async function kafkaJSToRdKafkaConfig(config) {
     }
     switch (setLevel) {
       case logLevel.NOTHING:
-        globalConfig["log_level"] = 0; /* LOG_EMERG - we don't have a true log nothing yet */
+        rdkafkaConfig["log_level"] = 0; /* LOG_EMERG - we don't have a true log nothing yet */
         break;
       case logLevel.ERROR:
-        globalConfig["log_level"] = 3 /* LOG_ERR */;
+        rdkafkaConfig["log_level"] = 3 /* LOG_ERR */;
         break;
       case logLevel.WARN:
-        globalConfig["log_level"] = 4 /* LOG_WARNING */;
+        rdkafkaConfig["log_level"] = 4 /* LOG_WARNING */;
         break;
       case logLevel.INFO:
-        globalConfig["log_level"] = 6 /* LOG_INFO */;
+        rdkafkaConfig["log_level"] = 6 /* LOG_INFO */;
         break;
       case logLevel.DEBUG:
-        globalConfig["debug"] = "all" /* this will set librdkafka log_level to 7 */;
+        rdkafkaConfig["debug"] = "all" /* Turn on debug logs for everything, otherwise this log level is not useful*/;
+        rdkafkaConfig["log_level"] = 7 /* LOG_DEBUG */;
         break;
       default:
         throw new error.KafkaJSError(CompatibilityErrorMessages.logLevelName(setLevel), {
@@ -324,20 +330,7 @@ async function kafkaJSToRdKafkaConfig(config) {
     }
   }
 
-  if (config.rdKafka) {
-    if (config.rdKafka.constructor === Function) {
-      await config.rdKafka(globalConfig, topicConfig);
-    } else {
-      Object.assign(globalConfig, config.rdKafka.globalConfig);
-      Object.assign(topicConfig, config.rdKafka.topicConfig);
-    }
-  }
-
-
-  if (!Object.hasOwn(globalConfig, 'log_level'))
-    globalConfig['log_level'] = Object.hasOwn(globalConfig, 'debug') ? 7 /* LOG_DEBUG */ : 6 /* LOG_INFO */;
-
-  return { globalConfig, topicConfig };
+  return rdkafkaConfig;
 }
 
 function checkAllowedKeys(allowedKeysSpecific, config) {
@@ -421,7 +414,6 @@ function createKafkaJsErrorFromLibRdKafkaError(librdKafkaError) {
     err = new error.KafkaJSError(librdKafkaError, properties);
   }
 
-  console.log("Converted err = " + JSON.stringify(err, null, 2) + " librdkafka erro = " + JSON.stringify(librdKafkaError, null, 2));
   return err;
 }