Add OAUTHBEARER support to promisified API

Author: milindl

MIGRATION.md

@@ -436,8 +436,12 @@ export interface IAdminClient {
     disconnect(): void;
 }
 
+export type EventHandlers = {
+    [event_key: string]: (...args: any[]) => void;
+};
+
 export abstract class AdminClient {
-    static create(conf: GlobalConfig): IAdminClient;
+    static create(conf: GlobalConfig, eventHandlers?: EventHandlers): IAdminClient;
 }
 
 export type RdKafka = {

index.d.ts

@@ -58,10 +58,19 @@ util.inherits(AdminClient, Client);
  * This is a factory method because it immediately starts an
  * active handle with the brokers.
  *
+ * @param {object} conf - Key value pairs to configure the admin client
+ * @param {object} eventHandlers - optional key value pairs of event handlers to attach to the client
+ *
  */
-function createAdminClient(conf) {
+function createAdminClient(conf, eventHandlers) {
   var client = new AdminClient(conf);
 
+  if (eventHandlers && typeof eventHandlers === 'object') {
+    for (const key in eventHandlers) {
+      client.on(key, eventHandlers[key]);
+    }
+  }
+
   // Wrap the error so we throw if it failed with some context
   LibrdKafkaError.wrap(client.connect(), true);
 
@@ -127,6 +136,7 @@ AdminClient.prototype.connect = function () {
   this._client.configureCallbacks(true, this._cb_configs);
   LibrdKafkaError.wrap(this._client.connect(), true);
   this._isConnected = true;
+  this.emit('ready', { name: this._client.name() });
 };
 
 /**

lib/admin.js

@@ -50,6 +50,12 @@ class Admin {
    */
   #logger = new DefaultLogger();
 
+  /**
+   * connectPromiseFunc is the set of promise functions used to resolve/reject the connect() promise.
+   * @type {{resolve: Function, reject: Function}|{}}
+   */
+  #connectPromiseFunc = null;
+
   /**
    * @constructor
    * @param {import("../../types/kafkajs").AdminConstructorConfig} config
@@ -95,6 +101,29 @@ class Admin {
     return rdKafkaConfig;
   }
 
+  #readyCb() {
+    if (this.#state !== AdminState.CONNECTING) {
+      /* The connectPromiseFunc might not be set, so we throw such an error. It's a state error that we can't recover from. Probably a bug. */
+      throw new error.KafkaJSError(`Ready callback called in invalid state ${this.#state}`, { code: error.ErrorCodes.ERR__STATE });
+    }
+    this.#state = AdminState.CONNECTED;
+
+    // Resolve the promise.
+    this.#connectPromiseFunc['resolve']();
+  }
+
+  /**
+   * Callback for the event.error event, either fails the initial connect(), or logs the error.
+   * @param {Error} err
+   */
+  #errorCb(err) {
+    if (this.#state === AdminState.CONNECTING) {
+      this.#connectPromiseFunc['reject'](err);
+    } else {
+      this.#logger.error(err);
+    }
+  }
+
   /**
    * Set up the client and connect to the bootstrap brokers.
    * @returns {Promise<void>} Resolves when connection is complete, rejects on error.
@@ -111,9 +140,11 @@ class Admin {
     return new Promise((resolve, reject) => {
       try {
         /* AdminClient creation is a synchronous operation for node-rdkafka */
-        this.#internalClient = RdKafka.AdminClient.create(config);
-        this.#state = AdminState.CONNECTED;
-        resolve();
+        this.#connectPromiseFunc = { resolve, reject };
+        this.#internalClient = RdKafka.AdminClient.create(config, {
+          'error': this.#errorCb.bind(this),
+          'ready': this.#readyCb.bind(this),
+        });
       } catch (err) {
         reject(createKafkaJsErrorFromLibRdKafkaError(err));
       }

lib/kafkajs/_admin.js

@@ -168,12 +168,12 @@ const CompatibilityErrorMessages = Object.freeze({
   brokerString: () =>
     "The 'brokers' property must be an array of strings.\n" +
     "For example: ['kafka:9092', 'kafka2:9093']\n",
-  saslOauthbearerUnsupported: () =>
-    "SASL mechanism OAUTHBEARER is not supported yet.",
   saslUnsupportedMechanism: (mechanism) =>
     `SASL mechanism ${mechanism} is not supported.`,
   saslUsernamePasswordString: (mechanism) =>
     `The 'sasl.username' and 'sasl.password' properties must be strings and must be present for the mechanism ${mechanism}.`,
+  saslOauthBearerProvider: () =>
+    `The 'oauthBearerProvider' property must be a function.`,
   sslObject: () =>
     "The 'ssl' property must be a boolean. Any additional configuration must be provided outside the kafkaJS block.\n" +
     "Before: \n" +
@@ -280,27 +280,65 @@ function kafkaJSToRdKafkaConfig(config) {
     const mechanism = sasl.mechanism.toUpperCase();
 
     if (mechanism === 'OAUTHBEARER') {
-      throw new error.KafkaJSError(CompatibilityErrorMessages.saslOauthbearerUnsupported(), {
-        code: error.ErrorCodes.ERR__NOT_IMPLEMENTED,
-      });
-    }
-
-    /* The mechanism must be PLAIN or SCRAM. */
-    if (mechanism !== 'PLAIN' && !mechanism.startsWith('SCRAM')) {
+      rdkafkaConfig["sasl.mechanism"] = mechanism;
+      if (Object.hasOwn(sasl, "oauthBearerProvider")) {
+        if (typeof sasl.oauthBearerProvider !== 'function') {
+          throw new error.KafkaJSError(CompatibilityErrorMessages.saslOauthBearerProvider(), {
+            code: error.ErrorCodes.ERR__INVALID_ARG,
+          });
+        }
+        rdkafkaConfig['oauthbearer_token_refresh_cb'] = function (oauthbearer_config) {
+          return sasl.oauthBearerProvider(oauthbearer_config)
+            .then((token) => {
+              if (!Object.hasOwn(token, 'value')) {
+                throw new error.KafkaJSError('Token must have a value property.', {
+                  code: error.ErrorCodes.ERR__INVALID_ARG,
+                });
+              } else if (!Object.hasOwn(token, 'principal')) {
+                throw new error.KafkaJSError('Token must have a principal property.', {
+                  code: error.ErrorCodes.ERR__INVALID_ARG,
+                });
+              } else if (!Object.hasOwn(token, 'lifetime')) {
+                throw new error.KafkaJSError('Token must have a lifetime property.', {
+                  code: error.ErrorCodes.ERR__INVALID_ARG,
+                });
+              }
+
+              // Recast token into a value expected by node-rdkafka's callback.
+              const setToken = {
+                tokenValue: token.value,
+                extensions: token.extensions,
+                principal: token.principal,
+                lifetime: token.lifetime,
+              };
+              return setToken;
+            })
+            .catch(err => {
+              if (!(err instanceof Error)) {
+                err = new Error(err);
+              }
+              throw err;
+            });
+        }
+      }
+    /* It's a valid case (unlike in KafkaJS) for oauthBearerProvider to be
+    * null, because librdkafka provides an unsecured token provider for
+    * non-prod usecases. So don't do anything in that case. */
+    } else if (mechanism === 'PLAIN' || mechanism.startsWith('SCRAM')) {
+      if (typeof sasl.username !== "string" || typeof sasl.password !== "string") {
+        throw new error.KafkaJSError(CompatibilityErrorMessages.saslUsernamePasswordString(mechanism), {
+          code: error.ErrorCodes.ERR__INVALID_ARG,
+        });
+      }
+      rdkafkaConfig["sasl.mechanism"] = mechanism;
+      rdkafkaConfig["sasl.username"] = sasl.username;
+      rdkafkaConfig["sasl.password"] = sasl.password;
+    } else {
       throw new error.KafkaJSError(CompatibilityErrorMessages.saslUnsupportedMechanism(mechanism), {
         code: error.ErrorCodes.ERR__INVALID_ARG,
       });
     }
 
-    if (typeof sasl.username !== "string" || typeof sasl.password !== "string") {
-      throw new error.KafkaJSError(CompatibilityErrorMessages.saslUsernamePasswordString(mechanism), {
-        code: error.ErrorCodes.ERR__INVALID_ARG,
-      });
-    }
-
-    rdkafkaConfig["sasl.mechanism"] = mechanism;
-    rdkafkaConfig["sasl.username"] = sasl.username;
-    rdkafkaConfig["sasl.password"] = sasl.password;
     withSASL = true;
   }
 

lib/kafkajs/_common.js

@@ -790,6 +790,7 @@ class Consumer {
     this.#state = ConsumerState.CONNECTING;
     this.#internalClient = new RdKafka.KafkaConsumer(rdKafkaConfig);
     this.#internalClient.on('ready', this.#readyCb.bind(this));
+    this.#internalClient.on('error', this.#errorCb.bind(this));
     this.#internalClient.on('event.error', this.#errorCb.bind(this));
     this.#internalClient.on('event.log', (msg) => loggerTrampoline(msg, this.#logger));
 

lib/kafkajs/_consumer.js

@@ -321,6 +321,7 @@ class Producer {
     this.#internalClient = new RdKafka.Producer(rdKafkaConfig);
     this.#internalClient.on('ready', this.#readyCb.bind(this));
     this.#internalClient.on('event.error', this.#errorCb.bind(this));
+    this.#internalClient.on('error', this.#errorCb.bind(this));
     this.#internalClient.on('event.log', (msg) => loggerTrampoline(msg, this.#logger));
 
     return new Promise((resolve, reject) => {
@@ -614,10 +615,10 @@ class Producer {
       throw new error.KafkaJSError(CompatibilityErrorMessages.sendOptionsAcks('sendBatch'), { code: error.ErrorCodes.ERR__INVALID_ARG });
     }
     if (Object.hasOwn(sendOptions, 'timeout')) {
-      throw new error.KafkaJSError(CompatibilityErrorMessages.sendOptionsTimeout('sendBatch'), { code: error.ErrorCodes.ERR__INVALID_ARG });
+      throw new error.KafkaJSError(CompatibilityErrorMessages.sendOptionsTimeout('timeout'), { code: error.ErrorCodes.ERR__INVALID_ARG });
     }
     if (Object.hasOwn(sendOptions, 'compression')) {
-      throw new error.KafkaJSError(CompatibilityErrorMessages.sendOptionsCompression('sendBatch'), { code: error.ErrorCodes.ERR__INVALID_ARG });
+      throw new error.KafkaJSError(CompatibilityErrorMessages.sendOptionsCompression('compression'), { code: error.ErrorCodes.ERR__INVALID_ARG });
     }
 
     if (sendOptions.topicMessages !== null && !Array.isArray(sendOptions.topicMessages)) {

lib/kafkajs/_producer.js

@@ -2,6 +2,7 @@
  * confluent-kafka-javascript - Node.js wrapper  for RdKafka C/C++ library
  *
  * Copyright (c) 2016-2023 Blizzard Entertainment
+ *           (c) 2024 Confluent, Inc.
  *
  * This software may be modified and distributed under the terms
  * of the MIT license.  See the LICENSE.txt file for details.
@@ -70,7 +71,7 @@ function Producer(conf, topicConf) {
   delete conf.dr_cb;
   delete conf.dr_msg_cb;
 
-  // client is an initialized consumer object
+  // client is an initialized producer object
   // @see NodeKafka::Producer::Init
   Client.call(this, conf, Kafka.Producer, topicConf);
 

lib/producer.js

@@ -105,6 +105,7 @@ void AdminClient::Init(v8::Local<v8::Object> exports) {
 
   // Inherited from NodeKafka::Connection
   Nan::SetPrototypeMethod(tpl, "configureCallbacks", NodeConfigureCallbacks);
+  Nan::SetPrototypeMethod(tpl, "name", NodeName);
 
   // Admin client operations
   Nan::SetPrototypeMethod(tpl, "createTopic", NodeCreateTopic);

src/admin.cc

@@ -116,18 +116,25 @@ RdKafka::TopicPartition* Connection::GetPartition(std::string &topic, int partit
   return RdKafka::TopicPartition::create(topic, partition);
 }
 
-bool Connection::IsConnected() {
+bool Connection::IsConnected() const {
   return !m_is_closing && m_client != NULL;
 }
 
-bool Connection::IsClosing() {
+bool Connection::IsClosing() const {
   return m_client != NULL && m_is_closing;
 }
 
 RdKafka::Handle* Connection::GetClient() {
   return m_client;
 }
 
+std::string Connection::Name() const {
+  if (!IsConnected()) {
+    return std::string("");
+  }
+  return std::string(m_client->name());
+}
+
 Baton Connection::CreateTopic(std::string topic_name) {
   return CreateTopic(topic_name, NULL);
 }
@@ -629,4 +636,10 @@ NAN_METHOD(Connection::NodeSetOAuthBearerTokenFailure) {
   info.GetReturnValue().Set(Nan::Null());
 }
 
+NAN_METHOD(Connection::NodeName) {
+  Connection* obj = ObjectWrap::Unwrap<Connection>(info.This());
+  std::string name = obj->Name();
+  info.GetReturnValue().Set(Nan::New(name).ToLocalChecked());
+}
+
 }  // namespace NodeKafka