Continue polling even when we don't receive any message, for example if all partitions are paused

Author: emasab

lib/kafkajs/_consumer.js

@@ -1374,17 +1374,21 @@ class Consumer {
     return ppc;
   }
 
+  #notifyNonEmpty() {
+    if (this.#nonEmpty) {
+      this.#nonEmpty.resolve();
+      this.#nonEmpty = null;
+    }
+    if (this.#messageCache)
+      this.#messageCache.notifyAvailablePartitions();
+  }
+
   #queueNonEmptyCb() {
     const nonEmptyAction = async () => {
       if (this.#fetchInProgress)
         await this.#fetchInProgress;
 
-      if (this.#nonEmpty) {
-        this.#nonEmpty.resolve();
-        this.#nonEmpty = null;
-      }
-      if (this.#messageCache)
-        this.#messageCache.notifyAvailablePartitions();
+      this.#notifyNonEmpty();
     };
     nonEmptyAction().catch((e) => {
       this.#logger.error(`Error in queueNonEmptyCb: ${e}`,
@@ -1457,19 +1461,32 @@ class Consumer {
    * @private
    */
   async #cacheExpirationLoop() {
+    const cacheExpirationInterval = BigInt(this.#cacheExpirationTimeoutMs * 1e6);
+    const maxFetchInterval = BigInt(1000 * 1e6);
     while (!this.#workerTerminationScheduled.resolved) {
       let now = hrtime.bigint();
-      const cacheExpiration = this.#lastFetchClockNs +
-        BigInt(this.#cacheExpirationTimeoutMs * 1e6);
+      const cacheExpirationTimeout = this.#lastFetchClockNs +
+        cacheExpirationInterval;
+      const maxFetchTimeout = this.#lastFetchClockNs +
+        maxFetchInterval;
 
-      if (now > cacheExpiration) {
+      if (now > cacheExpirationTimeout) {
         this.#addPendingOperation(() =>
           this.#clearCacheAndResetPositions());
         await this.#checkMaxPollIntervalNotExceeded(now);
         break;
       }
+      if (now > maxFetchTimeout) {
+        /* We need to continue fetching even when we're
+         * not getting any messages, for example when all partitions are
+         * paused. */
+        this.#notifyNonEmpty();
+      }
+
+      const awakeTime = maxFetchTimeout < cacheExpirationTimeout ?
+        maxFetchTimeout : cacheExpirationTimeout;
 
-      let interval = Number(cacheExpiration - now) / 1e6;
+      let interval = Number(awakeTime - now) / 1e6;
       if (interval < 100)
         interval = 100;
       await Timer.withTimeout(interval, this.#maxPollIntervalRestart);

test/promisified/admin/list_topics.spec.js

@@ -4,6 +4,7 @@ const {
     secureRandom,
     createTopic,
     createAdmin,
+    sleep,
 } = require('../testhelpers');
 const { ErrorCodes } = require('../../../lib').KafkaJS;
 
@@ -26,6 +27,10 @@ describe('Admin > listTopics', () => {
     it('should timeout', async () => {
         await admin.connect();
 
+        /* Await for the learned brokers to avoid having to do
+         * manual retries */
+        await sleep(1000);
+
         await expect(admin.listTopics({ timeout: 1 })).rejects.toHaveProperty(
             'code',
             ErrorCodes.ERR__TIMED_OUT