diff --git a/extensions/gc/GarbageCollector.js b/extensions/gc/GarbageCollector.js index 2e8e5700d..82c9a2b1b 100644 --- a/extensions/gc/GarbageCollector.js +++ b/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 ' + diff --git a/extensions/lifecycle/objectProcessor/LifecycleObjectTransitionProcessor.js b/extensions/lifecycle/objectProcessor/LifecycleObjectTransitionProcessor.js index cec526619..a9dc04f81 100644 --- a/extensions/lifecycle/objectProcessor/LifecycleObjectTransitionProcessor.js +++ b/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', }; }); diff --git a/package.json b/package.json index 0c93b6ff4..d0d4f248e 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "backbeat", - "version": "9.5.0-preview.8", + "version": "9.5.0-preview.9", "description": "Asynchronous queue and job manager", "main": "index.js", "scripts": { diff --git a/tests/unit/gc/GarbageCollector.spec.js b/tests/unit/gc/GarbageCollector.spec.js index 43106ec24..6202e0aff 100644 --- a/tests/unit/gc/GarbageCollector.spec.js +++ b/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'); diff --git a/tests/unit/lifecycle/LifecycleObjectExpirationProcessor.spec.js b/tests/unit/lifecycle/LifecycleObjectExpirationProcessor.spec.js index 9dda8aa95..cc4000aa6 100644 --- a/tests/unit/lifecycle/LifecycleObjectExpirationProcessor.spec.js +++ b/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]); diff --git a/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js b/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js index a0364415c..aa5138851 100644 --- a/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js +++ b/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]);