fixup! forward port workarround for queue populator batch getting stuck

bin/queuePopulator.js

@@ -51,58 +51,70 @@ function queueBatch(queuePopulator, taskState) {
 }
 /* eslint-enable no-param-reassign */
 
-const queuePopulator = new QueuePopulator(zkConfig, kafkaConfig,
-    qpConfig, httpsConfig, mConfig, rConfig, vConfig, extConfigs);
+function main() {
+    const queuePopulator = new QueuePopulator(zkConfig, kafkaConfig,
+        qpConfig, httpsConfig, mConfig, rConfig, vConfig, extConfigs);
 
-async.waterfall([
-    done => startProbeServer(qpConfig.probeServer, (err, probeServer) => {
+    async.waterfall([
+        done => startProbeServer(qpConfig.probeServer, (err, probeServer) => {
+            if (err) {
+                log.error('error starting probe server', {
+                    error: err,
+                    method: 'QueuePopulator::startProbeServer',
+                });
+                done(err);
+                return;
+            }
+            if (probeServer !== undefined) {
+                probeServer.addHandler([DEFAULT_LIVE_ROUTE, DEFAULT_READY_ROUTE],
+                    (res, log) => queuePopulator.handleLiveness(res, log)
+                );
+                probeServer.addHandler(DEFAULT_METRICS_ROUTE,
+                    (res, log) => queuePopulator.handleMetrics(res, log)
+                );
+            }
+            done();
+        }),
+        done => queuePopulator.open(done),
+        done => {
+            const taskState = {
+                batchInProgress: false,
+            };
+            schedule.scheduleJob(qpConfig.cronRule, () => {
+                queueBatch(queuePopulator, taskState);
+            });
+            done();
+        },
+    ], err => {
         if (err) {
-            log.error('error starting probe server', {
+            log.error('error during queue populator initialization', {
+                method: 'QueuePopulator::task',
                 error: err,
-                method: 'QueuePopulator::startProbeServer',
-            });
-            done(err);
-            return;
-        }
-        if (probeServer !== undefined) {
-            probeServer.addHandler([DEFAULT_LIVE_ROUTE, DEFAULT_READY_ROUTE],
-                (res, log) => queuePopulator.handleLiveness(res, log)
-            );
-            probeServer.addHandler(DEFAULT_METRICS_ROUTE,
-                (res, log) => queuePopulator.handleMetrics(res, log)
-            );
-        }
-        done();
-    }),
-    done => queuePopulator.open(done),
-    done => {
-        const taskState = {
-            batchInProgress: false,
-        };
-        schedule.scheduleJob(qpConfig.cronRule, () => {
-            queueBatch(queuePopulator, taskState);
-        });
-        done();
-    },
-], err => {
-    if (err) {
-        log.error('error during queue populator initialization', {
-            method: 'QueuePopulator::task',
-            error: err,
-        });
-        process.exit(1);
-    }
-});
-
-process.on('SIGTERM', () => {
-    log.info('received SIGTERM, exiting');
-    queuePopulator.close(error => {
-        if (error) {
-            log.error('failed to exit properly', {
-                error,
             });
             process.exit(1);
         }
-        process.exit(0);
     });
-});
+
+    process.on('SIGTERM', () => {
+        log.info('received SIGTERM, exiting');
+        queuePopulator.close(error => {
+            if (error) {
+                log.error('failed to exit properly', {
+                    error,
+                });
+                process.exit(1);
+            }
+            process.exit(0);
+        });
+    });
+}
+
+module.exports = {
+    queueBatch,
+};
+
+// run the main function if this script is run directly
+// otherwise the file is required as a module for testing
+if (require.main === module) {
+    main();
+}

tests/unit/bin/queuePopulator.spec.js

@@ -0,0 +1,49 @@
+const assert = require('assert');
+const sinon = require('sinon');
+
+const QueuePopulator = require('../../../lib/queuePopulator/QueuePopulator');
+const { queueBatch } = require('../../../bin/queuePopulator');
+
+describe('queuePopulator', () => {
+
+    afterEach(() => {
+        sinon.restore();
+    });
+
+    describe('processLogEntries', () => {
+        it('should reset batchInProgress if processLogEntries gets stuck', done => {
+            const qp = new QueuePopulator({}, {}, {}, {}, {}, {}, {}, {});
+            const processLogEntries = sinon.stub(qp, 'processLogEntries').callsFake(options => {
+                setTimeout(() => {
+                    options.onTimeout();
+                }, 1000);
+            });
+            const taskStatus = { batchInProgress: false };
+            queueBatch(qp, taskStatus);
+            assert.strictEqual(taskStatus.batchInProgress, true);
+            assert(processLogEntries.calledOnce);
+            setTimeout(() => {
+                assert.strictEqual(taskStatus.batchInProgress, false);
+                done();
+            }, 2000);
+        }).timeout(4000);
+
+        it('should reset batchInProgress when processLogEntries finishes', () => {
+            const qp = new QueuePopulator({}, {}, {}, {}, {}, {}, {}, {});
+            const processLogEntries = sinon.stub(qp, 'processLogEntries').yields();
+            const taskStatus = { batchInProgress: false };
+            queueBatch(qp, taskStatus);
+            assert(processLogEntries.calledOnce);
+            assert.strictEqual(taskStatus.batchInProgress, false);
+        });
+
+        it('should skip batch if one is currently in progress', () => {
+            const qp = new QueuePopulator({}, {}, {}, {}, {}, {}, {}, {});
+            const processLogEntries = sinon.stub(qp, 'processLogEntries').yields();
+            const taskStatus = { batchInProgress: true };
+            queueBatch(qp, taskStatus);
+            assert.strictEqual(taskStatus.batchInProgress, true);
+            assert(processLogEntries.notCalled);
+        });
+    });
+});

tests/unit/lib/queuePopulator/LogReader.spec.js

@@ -9,7 +9,6 @@ const { Logger } = require('werelogs');
 
 const LogReader = require('../../../../lib/queuePopulator/LogReader');
 
-
 class MockLogConsumer {
     constructor(params) {
         this.params = params || {};
@@ -40,6 +39,10 @@ describe('LogReader', () => {
         });
     });
 
+    afterEach(() => {
+        sinon.restore();
+    });
+
     // Currently the initial offset is set to 1 with mongodb backend,
     // it looks odd considering mongodb uses random IDs as log cursors
     // and could be cleaned up, but this test coming from 7.4 branch