rebalanceListener

Author: emasab

examples/kafkajs/consumer.js

@@ -2,18 +2,28 @@ const { Kafka } = require('../..').KafkaJS
 //const { Kafka } = require('kafkajs')
 
 async function consumerStart() {
+    let consumer;
     const kafka = new Kafka({
         brokers: ['pkc-8w6ry7.us-west-2.aws.devel.cpdev.cloud:9092'],
         ssl: true,
         sasl: {
             mechanism: 'plain',
         },
+        rebalanceListener: {
+          onPartitionsAssigned: async (assignment) => {
+            console.log(`Assigned partitions ${JSON.stringify(assignment)}`);
+          },
+          onPartitionsRevoked: async (assignment) => {
+            console.log(`Revoked partitions ${JSON.stringify(assignment)}`);
+            await consumer.commitOffsets();
+          }
+        },
         rdKafka: {
           'enable.auto.commit': false
         }
     });
 
-    const consumer = kafka.consumer({ groupId: 'test-group' });
+    consumer = kafka.consumer({ groupId: 'test-group' });
 
     await consumer.connect();
     console.log("Connected successfully");
@@ -50,7 +60,7 @@ async function consumerStart() {
     const disconnect = () => {
       process.off('SIGINT', disconnect);
       process.off('SIGTERM', disconnect);
-      consumer.disconnect().then(() => {
+      consumer.disconnect().finally(() => {
         console.log("Disconnected successfully");
       });
     }

lib/kafkajs/_consumer.js

@@ -29,10 +29,54 @@ class Consumer {
 
   async #finalizedConfig() {
     const config = await kafkaJSToRdKafkaConfig(this.#kJSConfig);
-    if (this.#kJSConfig.groupId != null) {
+    if (this.#kJSConfig.groupId) {
       config['group.id'] = this.#kJSConfig.groupId;
     }
     config['offset_commit_cb'] = true;
+    if (this.#kJSConfig.rebalanceListener) {
+      config['rebalance_cb'] = (err, assignment) => {
+        // Create the librdkafka error
+        err = LibrdKafkaError.create(err);
+
+        let call;
+        switch(err.code) {
+          case LibrdKafkaError.codes.ERR__ASSIGN_PARTITIONS:
+            call = (this.#kJSConfig.rebalanceListener.onPartitionsAssigned ?
+                   this.#kJSConfig.rebalanceListener.onPartitionsAssigned(assignment) :
+                   Promise.resolve()).catch(console.error);
+            break;
+          case LibrdKafkaError.codes.ERR__REVOKE_PARTITIONS:
+            call = (this.#kJSConfig.rebalanceListener.onPartitionsRevoked ?
+                    this.#kJSConfig.rebalanceListener.onPartitionsRevoked(assignment) :
+                    Promise.resolve()).catch(console.error);
+            break;
+          default:
+            call = Promise.reject().catch(() => {
+              console.error(`Unexpected rebalanceListener error code ${err.code}`);
+            });
+            break;
+        }
+
+        call
+        .finally(() => {
+          // Emit the event
+          this.#internalClient.emit('rebalance', err, assignment);
+
+          try {
+            if (err.code === LibrdKafkaError.codes.ERR__ASSIGN_PARTITIONS) {
+              this.#internalClient.assign(assignment);
+            } else {
+              this.#internalClient.unassign();
+            }
+          } catch (e) {
+            // Ignore exceptions if we are not connected
+            if (this.#internalClient.isConnected()) {
+              this.#internalClient.emit('rebalance.error', e);
+            }
+          }
+        });
+      };
+    }
     return config;
   }
 
@@ -164,7 +208,7 @@ class Consumer {
     }
   }
 
-  commitOffsets(topicPartitions = null) {
+  async commitOffsets(topicPartitions = null) {
     try {
       if (topicPartitions == null) {
         this.#internalClient.commitSync();
@@ -178,7 +222,6 @@ class Consumer {
         throw e;
       }
     }
-    return Promise.resolve()
   }
 
   seek(topicPartitionOffset) {