Seek implementation

Author: emasab

examples/kafkajs/consumer.js

@@ -36,6 +36,11 @@ async function consumerStart() {
         })
 
         if (++batch % 100 == 0) {
+          await consumer.seek({
+            topic,
+            partition,
+            offset: -2
+          });
           await consumer.commitOffsets();
           batch = 0;
         }

lib/kafkajs/_consumer.js

@@ -1,3 +1,4 @@
+const LibrdKafkaError = require('../error');
 const RdKafka = require('../rdkafka');
 const { kafkaJSToRdKafkaConfig } = require('./_common');
 
@@ -29,7 +30,7 @@ class Consumer {
   async #finalizedConfig() {
     const config = await kafkaJSToRdKafkaConfig(this.#kJSConfig);
     if (this.#kJSConfig.groupId != null) {
-      config["group.id"] = this.#kJSConfig.groupId;
+      config['group.id'] = this.#kJSConfig.groupId;
     }
     config['offset_commit_cb'] = true;
     return config;
@@ -43,20 +44,20 @@ class Consumer {
       this.#state = ConsumerState.CONNECTED;
 
       // Resolve the promise.
-      this.#connectPromiseFunc["resolve"]();
+      this.#connectPromiseFunc['resolve']();
   }
 
   #errorCb(args) {
       console.log('error', args);
       if (this.#state === ConsumerState.CONNECTING) {
-          this.#connectPromiseFunc["reject"](args);
+          this.#connectPromiseFunc['reject'](args);
       } else {
           // do nothing for now.
       }
   }
 
   #notImplemented() {
-    throw new Error("Not implemented");
+    throw new Error('Not implemented');
   }
 
   #createPayload(message) {
@@ -66,7 +67,7 @@ class Consumer {
     }
 
     let timestamp = message.timestamp ? new Date(message.timestamp).toISOString()
-                    : "";
+                    : '';
 
     var headers = undefined;
     if (message.headers) {
@@ -113,9 +114,17 @@ class Consumer {
     });
   }
 
+  #topicPartitionOffsetToRdKafka(tpo) {
+    return {
+      topic: tpo.topic,
+      partition: tpo.partition,
+      offset: Number(tpo.offset),
+    }
+  }
+
   async connect() {
       if (this.#state !== ConsumerState.INIT) {
-          return Promise.reject("Connect has already been called elsewhere.");
+          return Promise.reject('Connect has already been called elsewhere.');
       }
 
       this.#state = ConsumerState.CONNECTING;
@@ -126,9 +135,9 @@ class Consumer {
 
       return new Promise((resolve, reject) => {
           this.#connectPromiseFunc = {resolve, reject};
-          console.log("Connecting....");
+          console.log('Connecting....');
           this.#internalClient.connect();
-          console.log("connect() called");
+          console.log('connect() called');
       });
   }
 
@@ -142,7 +151,7 @@ class Consumer {
 
   async run(config) {
     if (this.#state !== ConsumerState.CONNECTED) {
-        throw new Error("Run must be called in state CONNECTED.");
+        throw new Error('Run must be called in state CONNECTED.');
     }
 
     while (this.#state === ConsumerState.CONNECTED) {
@@ -156,21 +165,34 @@ class Consumer {
   }
 
   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);
+    try {
+      if (topicPartitions == null) {
+        this.#internalClient.commitSync();
+      } else {
+        const topicPartitions = topicPartitions.map(
+          this.#topicPartitionOffsetToRdKafka);
+        this.#internalClient.commitSync(topicPartitions);
+      }
+    } catch (e) {
+      if (!e.code || e.code != LibrdKafkaError.codes.ERR__NO_OFFSET) {
+        throw e;
+      }
     }
     return Promise.resolve()
   }
 
   seek(topicPartitionOffset) {
-    this.#notImplemented();
+    return new Promise((resolve, reject) => {
+      const rdKafkaTopicPartitionOffset =
+      this.#topicPartitionOffsetToRdKafka(topicPartitionOffset);
+      this.#internalClient.seek(rdKafkaTopicPartitionOffset, 0, (err) => {
+        if (err) {
+          reject(new Error(`Seek error code ${err.code}`));
+        } else {
+          resolve();
+        }
+      });
+    }).catch(console.error); // Default handler
   }
 
   async describeGroup() {

types/kafkajs.d.ts

@@ -413,7 +413,7 @@ export type Consumer = {
   stop(): Promise<void>
   run(config?: ConsumerRunConfig): Promise<void>
   commitOffsets(topicPartitions: Array<TopicPartitionOffsetAndMetadata>): Promise<void>
-  seek(topicPartitionOffset: TopicPartitionOffset): void
+  seek(topicPartitionOffset: TopicPartitionOffset): Promise<void>
   describeGroup(): Promise<GroupDescription>
   pause(topics: Array<{ topic: string; partitions?: number[] }>): void
   paused(): TopicPartitions[]