rdKafka configuration

Author: emasab

examples/kafkajs/consumer.js

@@ -17,20 +17,13 @@ async function consumerStart() {
     await consumer.connect();
     console.log("Connected successfully");
 
-    const disconnect = () =>
-      consumer.disconnect().then(() => {
-        console.log("Disconnected successfully");
-      });
-    process.on('SIGINT', disconnect);
-    process.on('SIGTERM', disconnect);
-
-    consumer.subscribe({
+    await consumer.subscribe({
       topics: [
         "topic2"
       ]
     })
 
-    await consumer.run({
+    consumer.run({
       eachMessage: async ({ topic, partition, message }) => {
         console.log({
           topic,
@@ -41,6 +34,13 @@ async function consumerStart() {
         })
       },
     });
+
+    const disconnect = () =>
+      consumer.disconnect().then(() => {
+        console.log("Disconnected successfully");
+      });
+    process.on('SIGINT', disconnect);
+    process.on('SIGTERM', disconnect);
 }
 
 consumerStart()

lib/kafkajs/_common.js

@@ -1,4 +1,4 @@
-function kafkaJSToRdKafkaConfig(config) {
+async function kafkaJSToRdKafkaConfig(config) {
   const ret = {
     'allow.auto.create.topics': 'false'
   }
@@ -24,6 +24,14 @@ function kafkaJSToRdKafkaConfig(config) {
     ret['security.protocol'] = 'sasl_plaintext';
   }
 
+  if (config.rdKafka) {
+    if (config.rdKafka.constructor === Function) {
+      await config.rdKafka(ret);
+    } else {
+      Object.assign(ret, config.rdKafka);
+    }
+  }
+
   return ret;
 }
 

lib/kafkajs/_consumer.js

@@ -10,16 +10,28 @@ const ConsumerState = Object.freeze({
 });
 
 class Consumer {
-  #config = {}
+  #kJSConfig = null
+  #rdKafkaConfig = null;
   #internalClient = null;
   #connectPromiseFunc = {};
   #state = ConsumerState.INIT;
 
-  constructor(config) {
-    this.#config = kafkaJSToRdKafkaConfig(config);
-    if (config.groupId != null) {
-      this.#config["group.id"] = config.groupId;
+  constructor(kJSConfig) {
+    this.#kJSConfig = kJSConfig;
+  }
+
+  #config() {
+    if (!this.#rdKafkaConfig)
+      this.#rdKafkaConfig = this.#finalizedConfig();
+    return this.#rdKafkaConfig;
+  }
+
+  async #finalizedConfig() {
+    const config = await kafkaJSToRdKafkaConfig(this.#kJSConfig);
+    if (this.#kJSConfig.groupId != null) {
+      config["group.id"] = this.#kJSConfig.groupId;
     }
+    return config;
   }
 
   #readyCb(arg) {
@@ -98,13 +110,13 @@ class Consumer {
     });
   }
 
-  connect() {
+  async connect() {
       if (this.#state !== ConsumerState.INIT) {
           return Promise.reject("Connect has already been called elsewhere.");
       }
 
       this.#state = ConsumerState.CONNECTING;
-      this.#internalClient = new RdKafka.KafkaConsumer(this.#config);
+      this.#internalClient = new RdKafka.KafkaConsumer(await this.#config());
       this.#internalClient.on('ready', this.#readyCb.bind(this));
       this.#internalClient.on('event.error', this.#errorCb.bind(this));
       this.#internalClient.on('event.log', console.log);
@@ -117,11 +129,11 @@ class Consumer {
       });
   }
 
-  subscribe(subscription) {
+  async subscribe(subscription) {
     this.#internalClient.subscribe(subscription.topics);
   }
 
-  stop() {
+  async stop() {
     this.#notImplemented();
   }
 
@@ -140,15 +152,15 @@ class Consumer {
     }
   }
 
-  commitOffsets(topicPartitions) {
+  async commitOffsets(topicPartitions) {
     this.#notImplemented();
   }
 
   seek(topicPartitionOffset) {
     this.#notImplemented();
   }
 
-  describeGroup() {
+  async describeGroup() {
     this.#notImplemented();
   }
 

lib/kafkajs/_producer.js

@@ -10,14 +10,26 @@ const ProducerState = Object.freeze({
 });
 
 class Producer {
-  #config = {}
+  #kJSConfig = null
+  #rdKafkaConfig = null;
   #internalClient = null;
   #connectPromiseFunc = {};
   #state = ProducerState.INIT;
 
-  constructor(config) {
-      this.#config = kafkaJSToRdKafkaConfig(config);
-      this.#config.dr_cb = 'true';
+  constructor(kJSConfig) {
+    this.#kJSConfig = kJSConfig;
+  }
+
+  #config() {
+    if (!this.#rdKafkaConfig)
+      this.#rdKafkaConfig = this.#finalizedConfig();
+    return this.#rdKafkaConfig;
+  }
+
+  async #finalizedConfig() {
+    const config = await kafkaJSToRdKafkaConfig(this.#kJSConfig);
+    config.dr_cb = 'true';
+    return config;
   }
 
   #readyCb(arg) {
@@ -76,13 +88,13 @@ class Producer {
       }
   }
 
-  connect() {
+  async connect() {
       if (this.#state !== ProducerState.INIT) {
           return Promise.reject("Connect has already been called elsewhere.");
       }
 
       this.#state = ProducerState.CONNECTING;
-      this.#internalClient = new RdKafka.Producer(this.#config);
+      this.#internalClient = new RdKafka.Producer(await this.#config());
       this.#internalClient.on('ready', this.#readyCb.bind(this));
       this.#internalClient.on('event.error', this.#errorCb.bind(this));
       this.#internalClient.on('event.log', console.log);