Merge branch 'bugfix/BB-752' into q/9.1

Author: bert-e

extensions/gc/GarbageCollector.js

@@ -167,13 +167,17 @@ class GarbageCollector extends EventEmitter {
     }
 
     /**
-     * Close the lifecycle consumer
+     * Close the garbage collector consumer
      * @param {function} cb - callback function
      * @return {undefined}
      */
     close(cb) {
         this._logger.debug('closing garbage collector consumer');
-        this._consumer.close(cb);
+        if (this._consumer) {
+            this._consumer.close(cb);
+        } else {
+            cb();
+        }
     }
 
     processKafkaEntry(kafkaEntry, done) {

extensions/gc/service.js

@@ -100,7 +100,7 @@ initAndStart();
 
 process.on('SIGTERM', () => {
     logger.info('received SIGTERM, exiting');
-    garbageCollector.stop(() => {
+    garbageCollector.close(() => {
         process.exit(0);
     });
 });

extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js

@@ -145,7 +145,11 @@ class LifecycleObjectProcessor extends EventEmitter {
             clearInterval(this._deleteInactiveCredentialsInterval);
         }
 
-        this._consumer.close(cb);
+        if (this._consumers) {
+            this._consumers.close(cb);
+        } else {
+            cb();
+        }
     }
 
     /**

lib/BackbeatConsumer.js

@@ -823,6 +823,7 @@ class BackbeatConsumer extends EventEmitter {
         });
         // ensure consumer is active when calling offsetsStore() on
         // it, to avoid raising an exception (invalid state)
+        // TODO : potential issue here, see BB-758
         if (committableOffset !== null && !this.isPaused()) {
             this._consumer.offsetsStore([{ topic, partition,
                                            offset: committableOffset }]);

tests/unit/gc/GarbageCollector.spec.js

@@ -92,4 +92,47 @@ describe('garbage collector', function garbageCollector() {
               .setAttribute('target.owner', ownerId);
         gcTask.processActionEntry(action, done);
     });
+
+    describe('close() garbage collector', () => {
+        it('should call close on the consumer when it exists', done => {
+            const gcWithConsumer = new GarbageCollector({
+                kafkaConfig: {},
+                s3Config: { host: 'localhost', port: 7777 },
+                gcConfig: {
+                    topic: 'backbeat-gc',
+                    auth: { type: 'account', account: 'bart' },
+                    consumer: { groupId: 'backbeat-gc-consumer-group' },
+                },
+            });
+            let closeCalled = false;
+            gcWithConsumer._consumer = {
+                close: cb => {
+                    closeCalled = true;
+                    cb();
+                },
+            };
+            gcWithConsumer.close(err => {
+                assert.ifError(err);
+                assert.strictEqual(closeCalled, true);
+                done();
+            });
+        });
+
+        it('should call callback immediately when consumer is null', done => {
+            const gcNoConsumer = new GarbageCollector({
+                kafkaConfig: {},
+                s3Config: { host: 'localhost', port: 7777 },
+                gcConfig: {
+                    topic: 'backbeat-gc',
+                    auth: { type: 'account', account: 'bart' },
+                    consumer: { groupId: 'backbeat-gc-consumer-group' },
+                },
+            });
+            assert.strictEqual(gcNoConsumer._consumer, null);
+            gcNoConsumer.close(err => {
+                assert.ifError(err);
+                done();
+            });
+        });
+    });
 });

tests/unit/lifecycle/LifecycleObjectExpirationProcessor.spec.js

@@ -1,4 +1,5 @@
 const assert = require('assert');
+const sinon = require('sinon');
 const config = require('../../config.json');
 const LifecycleObjectExpirationProcessor =
     require('../../../extensions/lifecycle/objectProcessor/LifecycleObjectExpirationProcessor');
@@ -23,4 +24,41 @@ describe('LifecycleObjectExpirationProcessor', () => {
             config.extensions.lifecycle.objectTasksTopic,
         );
     });
+
+    describe('close() expiration processor', () => {
+        it('should call close on consumers when they exist', done => {
+            let closeCalled = false;
+            objectProcessor._consumers = {
+                close: cb => {
+                    closeCalled = true;
+                    cb();
+                },
+            };
+            objectProcessor.close(err => {
+                assert.ifError(err);
+                assert.strictEqual(closeCalled, true);
+                done();
+            });
+        });
+
+        it('should call callback immediately when consumers is null', done => {
+            assert.strictEqual(objectProcessor._consumers, null);
+            objectProcessor.close(err => {
+                assert.ifError(err);
+                done();
+            });
+        });
+
+        it('should clear deleteInactiveCredentialsInterval if set', done => {
+            const spy = sinon.spy(global, 'clearInterval');
+            const interval = setInterval(() => {}, 100000);
+            objectProcessor._deleteInactiveCredentialsInterval = interval;
+            objectProcessor.close(err => {
+                assert.ifError(err);
+                assert(spy.calledWith(interval));
+                spy.restore();
+                done();
+            });
+        });
+    });
 });