commitOffsets implementation

Author: emasab

examples/kafkajs/consumer.js

@@ -7,8 +7,9 @@ async function consumerStart() {
         ssl: true,
         sasl: {
             mechanism: 'plain',
-            username: '<fill>',
-            password: '<fill>',
+        },
+        rdKafka: {
+          'enable.auto.commit': false
         }
     });
 
@@ -23,6 +24,7 @@ async function consumerStart() {
       ]
     })
 
+    var batch = 0;
     consumer.run({
       eachMessage: async ({ topic, partition, message }) => {
         console.log({
@@ -32,13 +34,21 @@ async function consumerStart() {
           key: message.key?.toString(),
           value: message.value.toString(),
         })
+
+        if (++batch % 100 == 0) {
+          await consumer.commitOffsets();
+          batch = 0;
+        }
       },
     });
 
-    const disconnect = () =>
+    const disconnect = () => {
+      process.off('SIGINT', disconnect);
+      process.off('SIGTERM', disconnect);
       consumer.disconnect().then(() => {
         console.log("Disconnected successfully");
       });
+    }
     process.on('SIGINT', disconnect);
     process.on('SIGTERM', disconnect);
 }

lib/kafkajs/_consumer.js

@@ -31,6 +31,7 @@ class Consumer {
     if (this.#kJSConfig.groupId != null) {
       config["group.id"] = this.#kJSConfig.groupId;
     }
+    config['offset_commit_cb'] = true;
     return config;
   }
 
@@ -101,8 +102,10 @@ class Consumer {
   async #consumeSingle() {
     return new Promise((resolve, reject) => {
       this.#internalClient.consume(1, function(err, messages) {
-        if (err)
+        if (err) {
           reject(`Consume error code ${err.code}`);
+          return;
+        }
 
         const message = messages[0];
         resolve(message);
@@ -152,8 +155,18 @@ class Consumer {
     }
   }
 
-  async commitOffsets(topicPartitions) {
-    this.#notImplemented();
+  commitOffsets(topicPartitions = null) {
+    if (topicPartitions == null) {
+      this.#internalClient.commitSync();
+    } else {
+      const topicPartitions = topicPartitions.map((tpo) => ({
+        topic: tpo.topic,
+        partition: tpo.partition,
+        offset: Number(tpo.offset),
+      }))
+      this.#internalClient.commitSync(topicPartitions);
+    }
+    return Promise.resolve()
   }
 
   seek(topicPartitionOffset) {