Use separate topic for lifecycle transition tasks

conf/config.json

@@ -224,6 +224,7 @@
             "zookeeperPath": "/lifecycle",
             "bucketTasksTopic": "backbeat-lifecycle-bucket-tasks",
             "objectTasksTopic": "backbeat-lifecycle-object-tasks",
+            "transitionTasksTopic": "backbeat-lifecycle-transition-tasks",
             "conductor": {
                 "cronRule": "0 */5 * * * *",
                 "concurrency": 10,

extensions/lifecycle/LifecycleConfigValidator.js

@@ -12,6 +12,7 @@ const joiSchema = joi.object({
     zookeeperPath: joi.string().required(),
     bucketTasksTopic: joi.string().required(),
     objectTasksTopic: joi.string().required(),
+    transitionTasksTopic: joi.string().default(parent => parent.objectTasksTopic),
     coldStorageTopics: joi.array().items(joi.string()).unique().default([]),
     auth: authJoi.optional(),
     forceLegacyListing: joi.boolean().default(true),

extensions/lifecycle/bucketProcessor/LifecycleBucketProcessor.js

@@ -173,6 +173,7 @@ class LifecycleBucketProcessor {
             enabledRules: this._supportedRulesObject,
             bucketTasksTopic: this._lcConfig.bucketTasksTopic,
             objectTasksTopic: this._lcConfig.objectTasksTopic,
+            transitionTasksTopic: this._lcConfig.transitionTasksTopic,
             kafkaBacklogMetrics: this._kafkaBacklogMetrics,
             ncvHeap: this.ncvHeap,
             pausedLocations: this._pausedLocations,

extensions/lifecycle/objectProcessor/LifecycleObjectExpirationProcessor.js

@@ -22,6 +22,7 @@ class LifecycleObjectExpirationProcessor extends LifecycleObjectProcessor {
      * @param {Object} lcConfig - lifecycle configuration object
      * @param {String} lcConfig.auth - authentication info
      * @param {String} lcConfig.objectTasksTopic - lifecycle object topic name
+     * @param {String} lcConfig.transitionTasksTopic - lifecycle transition topic name
      * @param {Object} lcConfig.objectProcessor - kafka consumer object
      * @param {String} lcConfig.objectProcessor.groupId - kafka
      * consumer group id

extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js

@@ -41,8 +41,7 @@ class LifecycleObjectProcessor extends EventEmitter {
      * @param {Object} lcConfig - lifecycle configuration object
      * @param {String} lcConfig.auth - authentication info
      * @param {String} lcConfig.objectTasksTopic - lifecycle object topic name
-     * consumer group id
-     *  of max allowed concurrent operations
+     * @param {String} lcConfig.transitionTasksTopic - lifecycle transition topic name
      * @param {Object} s3Config - S3 configuration
      * @param {Object} s3Config.host - s3 endpoint host
      * @param {Number} s3Config.port - s3 endpoint port
@@ -73,7 +72,7 @@ class LifecycleObjectProcessor extends EventEmitter {
         return 'object-processor';
     }
 
-    _getObjecTaskConsumerParams() {
+    _getTopicConsumerParams(topic) {
         return {
             zookeeper: {
                 connectionString: this._zkConfig.connectionString,
@@ -83,7 +82,7 @@ class LifecycleObjectProcessor extends EventEmitter {
                 site: this._kafkaConfig.site,
                 backlogMetrics: this._kafkaConfig.backlogMetrics,
             },
-            topic: this._lcConfig.objectTasksTopic,
+            topic,
             groupId: this._processConfig.groupId,
             concurrency: this._processConfig.concurrency,
             queueProcessor: this.processObjectTaskEntry.bind(this),
@@ -94,9 +93,9 @@ class LifecycleObjectProcessor extends EventEmitter {
         };
     }
 
-    getConsumerParams() {
+    getConsumerParams(topic = this._lcConfig.objectTasksTopic) {
         return {
-            [this._lcConfig.objectTasksTopic]: this._getObjecTaskConsumerParams(),
+            [topic]: this._getTopicConsumerParams(topic),
         };
     }
 

extensions/lifecycle/objectProcessor/LifecycleObjectTransitionProcessor.js

@@ -29,6 +29,7 @@ class LifecycleObjectTransitionProcessor extends LifecycleObjectProcessor {
      * @param {Object} lcConfig - lifecycle configuration object
      * @param {String} lcConfig.auth - authentication info
      * @param {String} lcConfig.objectTasksTopic - lifecycle object topic name
+     * @param {String} lcConfig.transitionTasksTopic - lifecycle transition topic name
      * @param {Object} lcConfig.transitionProcessor - kafka consumer object
      * @param {String} lcConfig.transitionProcessor.groupId - kafka
      * consumer group id
@@ -49,7 +50,8 @@ class LifecycleObjectTransitionProcessor extends LifecycleObjectProcessor {
     }
 
     getConsumerParams() {
-        const consumerParams = super.getConsumerParams();
+        const consumerParams = super.getConsumerParams(this._lcConfig.transitionTasksTopic);
+
         const locations = require('../../../conf/locationConfig.json') || {};
 
         this._lcConfig.coldStorageTopics.forEach(topic => {

extensions/lifecycle/tasks/LifecycleTask.js

@@ -1129,7 +1129,7 @@ class LifecycleTask extends BackbeatTask {
             originLabel: 'lifecycle',
             fromLocation: objectMD.getDataStoreName(),
             contentLength: objectMD.getContentLength(),
-            resultsTopic: this.objectTasksTopic,
+            resultsTopic: this.transitionTasksTopic,
             accountId: params.accountId,
             attempt,
         });

tests/config.json

@@ -156,6 +156,7 @@
             "zookeeperPath": "/lifecycletest",
             "bucketTasksTopic": "backbeat-test-dummy-bucket-task",
             "objectTasksTopic": "backbeat-test-dummy-object-task",
+            "transitionTasksTopic": "backbeat-test-dummy-transition-task",
             "conductor": {
                 "cronRule": "0 */5 * * * *",
                 "probeServer": {
@@ -196,11 +197,29 @@
                     "port": 8554
                 }
             },
+            "transitionProcessor": {
+                "groupId": "backbeat-lifecycle-transition-processor-group",
+                "retry": {
+                    "maxRetries": 5,
+                    "timeoutS": 300,
+                    "backoff": {
+                        "min": 1000,
+                        "max": 300000,
+                        "jitter": 0.1,
+                        "factor": 1.5
+                    }
+                },
+                "concurrency": 10,
+                "probeServer": {
+                    "port": 8554
+                }
+            },
             "auth": {
                 "type": "account",
                 "account": "bart"
             },
-            "coldStorageArchiveTopicPrefix": "cold-archive-req-"
+            "coldStorageArchiveTopicPrefix": "cold-archive-req-",
+            "coldStorageTopics": []
         },
         "gc": {
             "topic": "backbeat-test-gc",

tests/functional/lifecycle/LifecycleConductor.spec.js

@@ -95,6 +95,7 @@ const baseLCConfig = {
     zookeeperPath: '/test/lifecycle',
     bucketTasksTopic,
     objectTasksTopic: 'backbeat-lifecycle-object-tasks-spec',
+    transitionTasksTopic: 'backbeat-lifecycle-transition-tasks-spec',
     conductor: {
         cronRule: '*/5 * * * * *',
         backlogControl: {

tests/functional/lifecycle/LifecycleTask.js

@@ -8,6 +8,12 @@ const LifecycleRule = require('arsenal').models.LifecycleRule;
 const LifecycleTask = require('../../../extensions/lifecycle/tasks/LifecycleTask');
 const testConfig = require('../../config.json');
 const { objectMD } = require('../utils/MetadataMock');
+const {
+    bucketTasksTopic,
+    objectTasksTopic,
+    transitionTasksTopic,
+} = require('./configObjects');
+
 const timeOptions = {
     expireOneDayEarlier: true,
     transitionOneDayEarlier: true,
@@ -328,10 +334,10 @@ class ProducerMock {
 
     sendToTopic(topicName, entries, cb) {
         const entry = JSON.parse(entries[0].message);
-        if (topicName === 'bucket-tasks') {
+        if (topicName === bucketTasksTopic) {
             this.sendCount.bucket++;
             this.entries.bucket.push(entry);
-        } else if (topicName === 'object-tasks') {
+        } else if (topicName === objectTasksTopic) {
             this.sendCount.object++;
             this.entries.object.push(entry.target.key);
         } else if (topicName === 'backbeat-data-mover') {
@@ -421,8 +427,9 @@ class LifecycleBucketProcessorMock {
             bootstrapList: [{ site: 'us-east-2', type: 'aws_s3' }],
             s3Endpoint: s3config.endpoint,
             s3Auth: lifecycle.auth,
-            bucketTasksTopic: 'bucket-tasks',
-            objectTasksTopic: 'object-tasks',
+            bucketTasksTopic,
+            objectTasksTopic,
+            transitionTasksTopic,
             kafkaBacklogMetrics: this._kafkaBacklogMetrics,
             pausedLocations: new Set(),
             log: this._log,
@@ -510,6 +517,7 @@ describe('lifecycle task functional tests', function dF() {
         entries.transitions.forEach(transition => {
             assert.strictEqual(transition.target.bucket, 'test-bucket');
             assert.strictEqual(transition.toLocation, 'us-east-2');
+            assert.strictEqual(transition.resultsTopic, transitionTasksTopic);
         });
     }
 

tests/functional/lifecycle/LifecycleTaskV2-versioned.js

@@ -39,9 +39,11 @@ const destinationLocation = 'us-east-2';
 
 const bucketTopic = 'bucket-topic';
 const objectTopic = 'object-topic';
+const transitionTopic = 'transition-topic';
 const dataMoverTopic = 'backbeat-data-mover';
 const testKafkaEntry = new TestKafkaEntry({
     objectTopic,
+    transitionTopic,
     bucketTopic,
     dataMoverTopic,
     ownerId,
@@ -72,6 +74,7 @@ describe('LifecycleTaskV2 with bucket versioned', () => {
                 producer,
                 bucketTasksTopic: bucketTopic,
                 objectTasksTopic: objectTopic,
+                transitionTasksTopic: transitionTopic,
                 kafkaBacklogMetrics: { snapshotTopicOffsets: () => {} },
                 pausedLocations: new Set(),
                 log,

tests/functional/lifecycle/LifecycleTaskV2.js

@@ -40,9 +40,11 @@ const destinationLocation = 'us-east-2';
 
 const bucketTopic = 'bucket-topic';
 const objectTopic = 'object-topic';
+const transitionTopic = 'transition-topic';
 const dataMoverTopic = 'backbeat-data-mover';
 const testKafkaEntry = new TestKafkaEntry({
     objectTopic,
+    transitionTopic,
     bucketTopic,
     dataMoverTopic,
     ownerId,
@@ -73,6 +75,7 @@ describe('LifecycleTaskV2 with bucket non-versioned', () => {
                 producer,
                 bucketTasksTopic: bucketTopic,
                 objectTasksTopic: objectTopic,
+                transitionTasksTopic: transitionTopic,
                 kafkaBacklogMetrics: { snapshotTopicOffsets: () => {} },
                 pausedLocations: new Set(),
                 log,

tests/functional/lifecycle/configObjects.js

@@ -1,5 +1,6 @@
 const bucketTasksTopic = 'bucket-tasks';
 const objectTasksTopic = 'object-tasks';
+const transitionTasksTopic = 'transition-tasks';
 
 const zkConfig = {
     connectionString: 'localhost:2181',
@@ -35,6 +36,7 @@ const lcConfig = {
     },
     bucketTasksTopic,
     objectTasksTopic,
+    transitionTasksTopic,
     rules: {
         expiration: {
             enabled: true,
@@ -69,6 +71,7 @@ const timeOptions = {
 module.exports = {
     bucketTasksTopic,
     objectTasksTopic,
+    transitionTasksTopic,
     zkConfig,
     kafkaConfig,
     lcConfig,

tests/functional/lifecycle/utils.js

@@ -6,6 +6,7 @@ class TestKafkaEntry {
     constructor(state) {
         const {
             objectTopic,
+            transitionTopic,
             bucketTopic,
             dataMoverTopic,
             ownerId,
@@ -14,6 +15,7 @@ class TestKafkaEntry {
         } = state;
 
         this.objectTopic = objectTopic;
+        this.transitionTopic = transitionTopic;
         this.bucketTopic = bucketTopic;
         this.dataMoverTopic = dataMoverTopic;
         this.ownerId = ownerId;
@@ -96,7 +98,7 @@ class TestKafkaEntry {
         assert.strictEqual(metrics.contentLength, contentLength);
 
         assert.strictEqual(message.toLocation, destinationLocation);
-        assert.strictEqual(message.resultsTopic, this.objectTopic);
+        assert.strictEqual(message.resultsTopic, this.transitionTopic);
     }
 
     expectBucketEntry(e, {

tests/unit/lifecycle/LifecycleObjectExpirationProcessor.spec.js

@@ -0,0 +1,26 @@
+const assert = require('assert');
+const config = require('../../config.json');
+const LifecycleObjectExpirationProcessor =
+    require('../../../extensions/lifecycle/objectProcessor/LifecycleObjectExpirationProcessor');
+
+describe('LifecycleObjectExpirationProcessor', () => {
+    let objectProcessor;
+
+    beforeEach(() => {
+        objectProcessor = new LifecycleObjectExpirationProcessor(
+            config.zookeeper,
+            config.kafka,
+            config.extensions.lifecycle,
+            config.s3,
+        );
+    });
+
+    it('should contain object tasks topic in consumer params', () => {
+        const consumerParams = objectProcessor.getConsumerParams();
+        assert.deepStrictEqual(Object.keys(consumerParams), [config.extensions.lifecycle.objectTasksTopic]);
+        assert.strictEqual(
+            consumerParams[config.extensions.lifecycle.objectTasksTopic].topic,
+            config.extensions.lifecycle.objectTasksTopic,
+        );
+    });
+});

tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js

@@ -0,0 +1,26 @@
+const assert = require('assert');
+const config = require('../../config.json');
+const LifecycleObjectTransitionProcessor =
+    require('../../../extensions/lifecycle/objectProcessor/LifecycleObjectTransitionProcessor');
+
+describe('LifecycleObjectExpirationProcessor', () => {
+    let objectProcessor;
+
+    beforeEach(() => {
+        objectProcessor = new LifecycleObjectTransitionProcessor(
+            config.zookeeper,
+            config.kafka,
+            config.extensions.lifecycle,
+            config.s3,
+        );
+    });
+
+    it('should contain transition tasks topic in consumer params', () => {
+        const consumerParams = objectProcessor.getConsumerParams();
+        assert.deepStrictEqual(Object.keys(consumerParams), [config.extensions.lifecycle.transitionTasksTopic]);
+        assert.strictEqual(
+            consumerParams[config.extensions.lifecycle.transitionTasksTopic].topic,
+            config.extensions.lifecycle.transitionTasksTopic
+        );
+    });
+});