Consume from the earliest offset where a lost message is unrecoverable

extensions/gc/GarbageCollector.js

@@ -121,14 +121,11 @@ class GarbageCollector extends EventEmitter {
     }
 
     /**
-     * Start kafka consumer. Emits a 'ready' event when
-     * consumer is ready.
-     *
-     * @return {undefined}
+     * Build the options of the garbage collector topic consumer
+     * @return {Object} consumer options
      */
-    start() {
-        let consumerReady = false;
-        this._consumer = new BackbeatConsumer({
+    _getConsumerOptions() {
+        return {
             kafka: {
                 hosts: this._kafkaConfig.hosts,
                 site: this._kafkaConfig.site,
@@ -143,7 +140,19 @@ class GarbageCollector extends EventEmitter {
             // sent to the gc don't have a key anyways)
             orderByFunc: null,
             queueProcessor: this.processKafkaEntry.bind(this),
-        });
+            fromOffset: 'earliest',
+        };
+    }
+
+    /**
+     * Start kafka consumer. Emits a 'ready' event when
+     * consumer is ready.
+     *
+     * @return {undefined}
+     */
+    start() {
+        let consumerReady = false;
+        this._consumer = new BackbeatConsumer(this._getConsumerOptions());
         this._consumer.on('error', () => {
             if (!consumerReady) {
                 this._logger.error('garbage collector failed to start the ' +

extensions/lifecycle/objectProcessor/LifecycleObjectTransitionProcessor.js

@@ -99,6 +99,8 @@ class LifecycleObjectTransitionProcessor extends LifecycleObjectProcessor {
     getConsumerParams() {
         const consumerParams = super.getConsumerParams(this._lcConfig.transitionTasksTopic);
 
+        consumerParams[this._lcConfig.transitionTasksTopic].fromOffset = 'earliest';
+
         const locations = require('../../../conf/locationConfig.json') || {};
 
         this._lcConfig.coldStorageTopics.forEach(topic => {
@@ -133,6 +135,7 @@ class LifecycleObjectTransitionProcessor extends LifecycleObjectProcessor {
                 maxQueued: this._processConfig.maxQueued,
                 queueProcessor: this.processColdStorageStatusEntry.bind(this),
                 circuitBreaker,
+                fromOffset: 'earliest',
             };
         });
 

tests/unit/gc/GarbageCollector.spec.js

@@ -64,6 +64,10 @@ describe('garbage collector', function garbageCollector() {
     after(() => {
         httpServer.close();
     });
+    it('should consume the gc topic from the earliest offset', () => {
+        assert.strictEqual(gc._getConsumerOptions().fromOffset, 'earliest');
+    });
+
     it('should skip unsupported action type', done => {
         expectBatchDeleteLocations = null;
         const action = ActionQueueEntry.create('foo');

tests/unit/lifecycle/LifecycleObjectExpirationProcessor.spec.js

@@ -16,6 +16,14 @@ describe('LifecycleObjectExpirationProcessor', () => {
         );
     });
 
+    it('should not pin fromOffset on the object tasks topic', () => {
+        const consumerParams = objectProcessor.getConsumerParams();
+        assert.strictEqual(
+            consumerParams[config.extensions.lifecycle.objectTasksTopic].fromOffset,
+            undefined,
+        );
+    });
+
     it('should contain object tasks topic in consumer params', () => {
         const consumerParams = objectProcessor.getConsumerParams();
         assert.deepStrictEqual(Object.keys(consumerParams), [config.extensions.lifecycle.objectTasksTopic]);

tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js

@@ -21,6 +21,26 @@ describe('LifecycleObjectTransitionProcessor', () => {
         );
     });
 
+    it('should consume the transition tasks topic from the earliest offset', () => {
+        const consumerParams = objectProcessor.getConsumerParams();
+        assert.strictEqual(
+            consumerParams[config.extensions.lifecycle.transitionTasksTopic].fromOffset,
+            'earliest'
+        );
+    });
+
+    it('should consume the cold status topics from the earliest offset', () => {
+        const coldTopic = `${config.extensions.lifecycle.coldStorageStatusTopicPrefix}location-dmf-v1`;
+        const coldProcessor = new LifecycleObjectTransitionProcessor(
+            config.zookeeper,
+            config.kafka,
+            { ...config.extensions.lifecycle, coldStorageTopics: [coldTopic] },
+            config.s3,
+        );
+        const consumerParams = coldProcessor.getConsumerParams();
+        assert.strictEqual(consumerParams[coldTopic].fromOffset, 'earliest');
+    });
+
     it('should contain transition tasks topic in consumer params', () => {
         const consumerParams = objectProcessor.getConsumerParams();
         assert.deepStrictEqual(Object.keys(consumerParams), [config.extensions.lifecycle.transitionTasksTopic]);