forward port workarround for queue populator batch getting stuck

original commit hash: dfcc2f5

Issue: BB-526

bin/queuePopulator.js

@@ -27,11 +27,16 @@ function queueBatch(queuePopulator, taskState) {
         log.debug('skipping batch: previous one still in progress');
         return undefined;
     }
+    const onTimeout = () => {
+        // reset the flag to allow a new batch to start in case the
+        // previous batch timed out
+        taskState.batchInProgress = false;
+    };
     log.debug('start queueing batch');
     taskState.batchInProgress = true;
     const maxRead = qpConfig.batchMaxRead;
     const timeoutMs = qpConfig.batchTimeoutMs;
-    queuePopulator.processLogEntries({ maxRead, timeoutMs }, err => {
+    queuePopulator.processLogEntries({ maxRead, timeoutMs, onTimeout }, err => {
         taskState.batchInProgress = false;
         if (err) {
             log.error('an error occurred during batch processing', {

lib/queuePopulator/LogReader.js

@@ -52,6 +52,8 @@ class LogReader {
         this._metricsHandler = params.metricsHandler;
         // TODO: use a common handler for zk metrics from all extensions
         this._zkMetricsHandler = params.zkMetricsHandler;
+
+        this.batchTimeoutSeconds = parseInt(process.env.BATCH_TIMEOUT_SECONDS, 10) || 300;
     }
 
     _setEntryBatch(entryBatch) {
@@ -291,6 +293,14 @@ class LogReader {
             timeoutMs: params.timeoutMs,
             logger: this.log.newRequestLogger(),
         };
+        const batchTimeoutTimer = setTimeout(() => {
+            this.log.error('queue populator batch timeout', {
+                logStats: batchState.logStats,
+            });
+            if (params.onTimeout) {
+                params.onTimeout();
+            }
+        },  this.batchTimeoutSeconds * 1000);
         async.waterfall([
             next => this._processReadRecords(params, batchState, next),
             next => this._processPrepareEntries(batchState, next),
@@ -299,6 +309,7 @@ class LogReader {
             next => this._processSaveLogOffset(batchState, next),
         ],
             err => {
+                clearTimeout(batchTimeoutTimer);
                 if (err) {
                     return done(err);
                 }

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

@@ -244,4 +244,20 @@ describe('LogReader', () => {
             });
         });
     });
+
+    describe('processLogEntries', () => {
+        it('should call function when the timeout is hit', done => {
+            logReader.batchTimeoutSeconds = 1;
+            // logReader will become stuck as _processReadRecords will never
+            // call the callback
+            sinon.stub(logReader, '_processReadRecords').returns();
+            const onTimeout = sinon.stub().returns();
+            logReader.processLogEntries({ onTimeout }, () => {});
+            setTimeout(() => {
+                delete process.env.BATCH_TIMEOUT_SECONDS;
+                assert(onTimeout.calledOnce);
+                done();
+            }, 2000);
+        }).timeout(4000);
+    });
 });