Batch max size is whatever available in the queue for that partition

Author: emasab

lib/kafkajs/_consumer.js

@@ -155,17 +155,6 @@ class Consumer {
    */
   #messageCache = null;
 
-  /**
-   * The maximum size of the message cache.
-   * Will be adjusted dynamically.
-   */
-  #messageCacheMaxSize = 1;
-
-  /**
-   * Number of times we tried to increase the cache.
-   */
-  #increaseCount = 0;
-
   /**
    * Whether the user has enabled manual offset management (commits).
    */
@@ -311,8 +300,6 @@ class Consumer {
      * consumed messages upto N from the internalClient, but the user has stale'd the cache
      * after consuming just k (< N) messages. We seek back to last consumed offset + 1. */
     this.#messageCache.clear();
-    this.#messageCacheMaxSize = 1;
-    this.#increaseCount = 0;
     const clearPartitions = this.assignment();
     const seeks = [];
     for (const topicPartition of clearPartitions) {
@@ -844,33 +831,6 @@ class Consumer {
     await this.commitOffsets();
   }
 
-  /**
-   * Request a size increase.
-   * It increases the size by 2x, but only if the size is less than 1024,
-   * only if the size has been requested to be increased twice in a row.
-   * @private
-   */
-   #increaseMaxSize() {
-     if (this.#messageCacheMaxSize === 1024)
-         return;
-     this.#increaseCount++;
-     if (this.#increaseCount <= 1)
-         return;
-     this.#messageCacheMaxSize = Math.min(this.#messageCacheMaxSize << 1, 1024);
-     this.#increaseCount = 0;
-  }
-
-  /**
-   * Request a size decrease.
-   * It decreases the size to 80% of the last received size, with a minimum of 1.
-   * @param {number} recvdSize - the number of messages received in the last poll.
-   * @private
-   */
-  #decreaseMaxSize(recvdSize) {
-      this.#messageCacheMaxSize = Math.max(Math.floor((recvdSize * 8) / 10), 1);
-      this.#increaseCount = 0;
-  }
-
   /**
    * Converts a list of messages returned by node-rdkafka into a message that can be used by the eachBatch callback.
    * @param {import("../..").Message[]} messages - must not be empty. Must contain messages from the same topic and partition.
@@ -1001,11 +961,6 @@ class Consumer {
       const res = takeFromCache();
       this.#lastFetchClockNs = hrtime.bigint();
       this.#maxPollIntervalRestart.resolve();
-      if (messages.length === this.#messageCacheMaxSize) {
-        this.#increaseMaxSize();
-      } else {
-        this.#decreaseMaxSize(messages.length);
-      }
       return res;
     } finally {
       this.#fetchInProgress.resolve();
@@ -1040,7 +995,7 @@ class Consumer {
     }
 
     return this.#fetchAndResolveWith(() => this.#messageCache.next(),
-      this.#messageCacheMaxSize);
+      Number.MAX_SAFE_INTEGER);
   }
 
   /**
@@ -1071,7 +1026,7 @@ class Consumer {
 
     return this.#fetchAndResolveWith(() =>
         this.#messageCache.nextN(null, size),
-      this.#messageCacheMaxSize);
+      Number.MAX_SAFE_INTEGER);
   }
 
   /**
@@ -1560,7 +1515,7 @@ class Consumer {
     this.#concurrency = config.partitionsConsumedConcurrently;
     const perMessageProcessor = config.eachMessage ? this.#messageProcessor : this.#batchProcessor;
     /* TODO: make this dynamic, based on max batch size / size of last message seen. */
-    const maxBatchSize = 32;
+    const maxBatchSize = -1;
     const fetcher = config.eachMessage
       ? (savedIdx) => this.#consumeSingleCached(savedIdx)
       : (savedIdx) => this.#consumeCachedN(savedIdx, maxBatchSize);