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

@@ -11,6 +11,8 @@ const { metricsExtension, metricsTypeQueued } =
     require('../../extensions/replication/constants');
 const { transformKey } = require('../util/entry');
 
+const BATCH_TIMEOUT_SECONDS = 300;
+
 class LogReader {
 
     /**
@@ -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();
+            }
+        }, BATCH_TIMEOUT_SECONDS * 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);
                 }