Add setSaslCredentials for producer

Author: milindl

index.d.ts

@@ -198,6 +198,8 @@ export abstract class Client<Events extends string> extends EventEmitter {
     queryWatermarkOffsets(topic: string, partition: number, timeout: number, cb?: (err: LibrdKafkaError, offsets: WatermarkOffsets) => any): any;
     queryWatermarkOffsets(topic: string, partition: number, cb?: (err: LibrdKafkaError, offsets: WatermarkOffsets) => any): any;
 
+    setSaslCredentials(username: string, password: string): void;
+
     on<E extends Events>(event: E, listener: EventListener<E>): this;
     once<E extends Events>(event: E, listener: EventListener<E>): this;
 }

lib/client.js

@@ -466,6 +466,21 @@ Client.prototype.offsetsForTimes = function(toppars, timeout, cb) {
   });
 };
 
+/**
+ * Change SASL credentials to be sent on the next authentication attempt.
+ *
+ * @param {string} username
+ * @param {string} password
+ * @note Only applicable if SASL authentication is being used.
+ */
+Client.prototype.setSaslCredentials = function(username, password) {
+  if (!this.isConnected()) {
+      return;
+  }
+
+  this._client.setSaslCredentials(username, password);
+};
+
 /**
  * Wrap a potential RdKafka error.
  *

lib/kafkajs/_producer.js

@@ -645,6 +645,40 @@ class Producer {
   logger() {
     return this.#logger;
   }
+
+  /**
+   * Change SASL credentials to be sent on the next authentication attempt.
+   *
+   * @param {string} args.username
+   * @param {string} args.password
+   * @note Only applicable if SASL authentication is being used.
+   */
+  setSaslCredentials(args = {}) {
+    if (!Object.hasOwn(args, 'username')) {
+      throw new error.KafkaJSError("username must be set for setSaslCredentials", { code: error.ErrorCodes.ERR__INVALID_ARG });
+    }
+
+    if (!Object.hasOwn(args, 'password')) {
+      throw new error.KafkaJSError("password must be set for setSaslCredentials", { code: error.ErrorCodes.ERR__INVALID_ARG });
+    }
+
+    /**
+     * In case we've not started connecting yet, just modify the configuration for
+     * the first connection attempt.
+     */
+    if (this.#state < ProducerState.CONNECTING) {
+      this.#userConfig['sasl.username'] = args.username;
+      this.#userConfig['sasl.password'] = args.password;
+      if (Object.hasOwn(this.#userConfig, 'kafkaJS') && Object.hasOwn(this.#userConfig.kafkaJS, 'sasl')) {
+        this.#userConfig.kafkaJS.sasl.username = args.username;
+        this.#userConfig.kafkaJS.sasl.password = args.password;
+      }
+      return;
+    }
+
+
+    this.#internalClient.setSaslCredentials(args.username, args.password);
+  }
 }
 
 module.exports = { Producer, CompressionTypes };

src/admin.cc

@@ -88,6 +88,7 @@ void AdminClient::Init(v8::Local<v8::Object> exports) {
 
   Nan::SetPrototypeMethod(tpl, "connect", NodeConnect);
   Nan::SetPrototypeMethod(tpl, "disconnect", NodeDisconnect);
+  Nan::SetPrototypeMethod(tpl, "setSaslCredentials", NodeSetSaslCredentials);
 
   constructor.Reset(
     (tpl->GetFunction(Nan::GetCurrentContext())).ToLocalChecked());

src/connection.cc

@@ -67,6 +67,18 @@ Connection::~Connection() {
   }
 }
 
+Baton Connection::rdkafkaErrorToBaton(RdKafka::Error* error) {
+  if ( NULL == error) {
+    return Baton(RdKafka::ERR_NO_ERROR);
+  }
+  else {
+    Baton result(error->code(), error->str(), error->is_fatal(),
+                 error->is_retriable(), error->txn_requires_abort());
+    delete error;
+    return result;
+  }
+}
+
 RdKafka::TopicPartition* Connection::GetPartition(std::string &topic) {
   return RdKafka::TopicPartition::create(topic, RdKafka::Topic::PARTITION_UA);
 }
@@ -215,6 +227,25 @@ Baton Connection::GetMetadata(
   }
 }
 
+Baton Connection::SetSaslCredentials(
+  std::string username, std::string password) {
+  RdKafka::Error *error;
+
+  if (IsConnected()) {
+    scoped_shared_read_lock lock(m_connection_lock);
+    if (IsConnected()) {
+      // Always send true - we
+      error = m_client->sasl_set_credentials(username, password);
+    } else {
+      return Baton(RdKafka::ERR__STATE);
+    }
+  } else {
+    return Baton(RdKafka::ERR__STATE);
+  }
+
+  return rdkafkaErrorToBaton(error);
+}
+
 void Connection::ConfigureCallback(const std::string &string_key, const v8::Local<v8::Function> &cb, bool add) {
   if (string_key.compare("event_cb") == 0) {
     if (add) {
@@ -337,6 +368,39 @@ NAN_METHOD(Connection::NodeQueryWatermarkOffsets) {
   info.GetReturnValue().Set(Nan::Null());
 }
 
+NAN_METHOD(Connection::NodeSetSaslCredentials) {
+  if (!info[0]->IsString()) {
+    Nan::ThrowError("1st parameter must be a username string");
+    return;
+  }
+
+  if (!info[1]->IsString()) {
+    Nan::ThrowError("2nd parameter must be a password string");
+    return;
+  }
+
+  // Get string pointer for the username
+  Nan::Utf8String usernameUTF8(Nan::To<v8::String>(info[0]).ToLocalChecked());
+  // The first parameter is the username
+  std::string username(*usernameUTF8);
+
+  // Get string pointer for the password
+  Nan::Utf8String passwordUTF8(Nan::To<v8::String>(info[1]).ToLocalChecked());
+  // The first parameter is the password
+  std::string password(*passwordUTF8);
+
+  Connection* obj = ObjectWrap::Unwrap<Connection>(info.This());
+  Baton b = obj->SetSaslCredentials(username, password);
+
+  if (b.err() != RdKafka::ERR_NO_ERROR) {
+    v8::Local<v8::Value> errorObject = b.ToObject();
+    return Nan::ThrowError(errorObject);
+  }
+
+  info.GetReturnValue().Set(Nan::Null());
+}
+
+
 // Node methods
 NAN_METHOD(Connection::NodeConfigureCallbacks) {
   Nan::HandleScope scope;

src/connection.h

@@ -55,6 +55,7 @@ class Connection : public Nan::ObjectWrap {
   Baton GetMetadata(bool, std::string, int);
   Baton QueryWatermarkOffsets(std::string, int32_t, int64_t*, int64_t*, int);
   Baton OffsetsForTimes(std::vector<RdKafka::TopicPartition*> &, int);
+  Baton SetSaslCredentials(std::string, std::string);
 
   RdKafka::Handle* GetClient();
 
@@ -74,6 +75,7 @@ class Connection : public Nan::ObjectWrap {
 
   static Nan::Persistent<v8::Function> constructor;
   static void New(const Nan::FunctionCallbackInfo<v8::Value>& info);
+  Baton rdkafkaErrorToBaton(RdKafka::Error* error);
 
   bool m_has_been_disconnected;
   bool m_is_closing;
@@ -90,6 +92,7 @@ class Connection : public Nan::ObjectWrap {
   static NAN_METHOD(NodeGetMetadata);
   static NAN_METHOD(NodeQueryWatermarkOffsets);
   static NAN_METHOD(NodeOffsetsForTimes);
+  static NAN_METHOD(NodeSetSaslCredentials);
 };
 
 }  // namespace NodeKafka

src/kafka-consumer.cc

@@ -504,6 +504,7 @@ void KafkaConsumer::Init(v8::Local<v8::Object> exports) {
   Nan::SetPrototypeMethod(tpl, "queryWatermarkOffsets", NodeQueryWatermarkOffsets);  // NOLINT
   Nan::SetPrototypeMethod(tpl, "offsetsForTimes", NodeOffsetsForTimes);
   Nan::SetPrototypeMethod(tpl, "getWatermarkOffsets", NodeGetWatermarkOffsets);
+  Nan::SetPrototypeMethod(tpl, "setSaslCredentials", NodeSetSaslCredentials);
 
   /*
    * @brief Methods exposed to do with message retrieval

src/producer.cc

@@ -71,6 +71,7 @@ void Producer::Init(v8::Local<v8::Object> exports) {
   Nan::SetPrototypeMethod(tpl, "getMetadata", NodeGetMetadata);
   Nan::SetPrototypeMethod(tpl, "queryWatermarkOffsets", NodeQueryWatermarkOffsets);  // NOLINT
   Nan::SetPrototypeMethod(tpl, "poll", NodePoll);
+  Nan::SetPrototypeMethod(tpl, "setSaslCredentials", NodeSetSaslCredentials);
 
   /*
    * @brief Methods exposed to do with message production
@@ -351,18 +352,6 @@ void Producer::ConfigureCallback(const std::string &string_key, const v8::Local<
   }
 }
 
-Baton rdkafkaErrorToBaton(RdKafka::Error* error) {
-  if ( NULL == error) {
-    return Baton(RdKafka::ERR_NO_ERROR);
-  }
-  else {
-    Baton result(error->code(), error->str(), error->is_fatal(),
-                 error->is_retriable(), error->txn_requires_abort());
-    delete error;
-    return result;
-  }
-}
-
 Baton Producer::InitTransactions(int32_t timeout_ms) {
   if (!IsConnected()) {
     return Baton(RdKafka::ERR__STATE);