Update Admin Client for config changes

Author: milindl

examples/kafkajs/admin.js

@@ -3,7 +3,9 @@ const { Kafka } = require('../..').KafkaJS
 
 async function adminStart() {
   const kafka = new Kafka({
-    brokers: ['<fill>'],
+    kafkaJs: {
+      brokers: ['localhost:9092'],
+    }
   });
 
   const admin = kafka.admin();

lib/kafkajs/_admin.js

@@ -1,5 +1,8 @@
 const RdKafka = require('../rdkafka');
-const { kafkaJSToRdKafkaConfig, createKafkaJsErrorFromLibRdKafkaError } = require('./_common');
+const { kafkaJSToRdKafkaConfig,
+  createKafkaJsErrorFromLibRdKafkaError,
+  DefaultLogger,
+  checkAllowedKeys } = require('./_common');
 const error = require('./_error');
 
 /**
@@ -17,16 +20,16 @@ const AdminState = Object.freeze({
 
 class Admin {
   /**
-   * kJSConfig is the merged kafkaJS config object.
-   * @type {import("../../types/kafkajs").AdminConfig  & import("../../types/kafkajs").KafkaConfig}
+   * The config supplied by the user.
+   * @type {import("../../types/kafkajs").AdminConstructorConfig|null}
    */
-  #kJSConfig = null;
+  #userConfig = null;
 
   /**
-   * rdKafkaConfig contains the config objects that will be passed to node-rdkafka.
-   * @type {{globalConfig: import("../../types/config").GlobalConfig}|null}
+   * The config realized after processing any compatibility options.
+   * @type {import("../../types/config").GlobalConfig|null}
    */
-  #rdKafkaConfig = null;
+  #internalConfig = null;
 
   /**
    * internalClient is the node-rdkafka client used by the API.
@@ -39,25 +42,55 @@ class Admin {
    */
   #state = AdminState.INIT;
 
+  /**
+   * A logger for the admin client.
+   * @type {import("../../types/kafkajs").Logger}
+   */
+  #logger = new DefaultLogger();
+
   /**
    * @constructor
-   * @param {import("../../types/kafkajs").ProducerConfig} kJSConfig
+   * @param {import("../../types/kafkajs").AdminConstructorConfig} config
    */
-  constructor(kJSConfig) {
-    this.#kJSConfig = kJSConfig;
+  constructor(config) {
+    this.#userConfig = config;
   }
 
-  async #config() {
-    if (!this.#rdKafkaConfig)
-      this.#rdKafkaConfig = await this.#finalizedConfig();
-    return this.#rdKafkaConfig;
+  #config() {
+    if (!this.#internalConfig)
+      this.#internalConfig = this.#finalizedConfig();
+    return this.#internalConfig;
   }
 
-  async #finalizedConfig() {
-    /* This sets the common configuration options for the client. */
-    const { globalConfig } = await kafkaJSToRdKafkaConfig(this.#kJSConfig);
+  #kafkaJSToAdminConfig(kjsConfig) {
+    if (!kjsConfig || Object.keys(kjsConfig).length === 0) {
+      return {};
+    }
+
+    const disallowedKey = checkAllowedKeys('admin', kjsConfig);
+    if (disallowedKey) {
+      throw new error.KafkaJSError(CompatibilityErrorMessages.unsupportedKey(disallowedKey), { code: error.ErrorCodes.ERR__INVALID_ARG });
+    }
+
+    const rdKafkaConfig = kafkaJSToRdKafkaConfig(kjsConfig);
+    return rdKafkaConfig;
+  }
+
+  #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) {
+      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 */
+    delete rdKafkaConfig.kafkaJs;
 
-    return { globalConfig };
+    return rdKafkaConfig;
   }
 
   /**
@@ -71,12 +104,12 @@ class Admin {
 
     this.#state = AdminState.CONNECTING;
 
-    const { globalConfig } = await this.#config();
+     const config = this.#config();
 
     return new Promise((resolve, reject) => {
       try {
         /* AdminClient creation is a synchronous operation for node-rdkafka */
-        this.#internalClient = RdKafka.AdminClient.create(globalConfig);
+        this.#internalClient = RdKafka.AdminClient.create(config);
         this.#state = AdminState.CONNECTED;
         resolve();
       } catch (err) {

lib/kafkajs/_common.js

@@ -53,7 +53,8 @@ const kafkaJsProperties = {
     'autoCommitInterval',
     'autoCommitThreshold',
     'rebalanceListener',
-  ]
+  ],
+  admin: [],
 }
 
 const logLevel = Object.freeze({

lib/kafkajs/_producer.js

@@ -84,7 +84,6 @@ class Producer {
   }
 
   #config() {
-    this.#internalClient
     if (!this.#internalConfig)
       this.#internalConfig = this.#finalizedConfig();
     return this.#internalConfig;

types/kafkajs.d.ts

@@ -176,6 +176,10 @@ export interface AdminConfig {
   retry?: RetryOptions
 }
 
+export interface AdminConstructorConfig extends GlobalConfig {
+  kafkaJs?: AdminConfig;
+}
+
 export interface ITopicConfig {
   topic: string
   numPartitions?: number