fix: offset subscription for ACS during startup

restorecommerce · Mar 22, 2024 · 00b3aa0 · 00b3aa0
1 parent f4555b9
commit 00b3aa0
Showing 1 changed file with 2 additions and 1 deletion.
diff --git a/src/worker.ts b/src/worker.ts
@@ -305,9 +305,10 @@ export class Worker {
       const topicCfg = kafkaConfig.topics[topicLabel];
       const topic = await events.topic(topicCfg.topic);
       const offSetValue = await this.offsetStore.getOffset(topicCfg.topic);
+      logger.info('subscribing to topic with offset value', topicCfg.topic, offSetValue);
       if (topicCfg.events) {
         for (let eventName of topicCfg.events) {
-          await topic.on(eventName, eventListener);
+          await topic.on(eventName, eventListener, { startingOffset: offSetValue });
         }
       }
     }