diff --git a/CHANGELOG.md b/CHANGELOG.md
index b5823789..f6e5830a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,16 @@
+# confluent-kafka-javascript 1.6.1
+
+v1.6.1 is a maintenance release. It is supported for all usage.
+
+### Enhancements
+
+1. Configurable batch size through the `js.consumer.max.batch.size` property 
+   and cache size through the `js.consumer.max.cache.size.per.worker.ms` 
+   property (#393).
+2. Fix for at-least-once guarantee not ensured in case a seek happens on one
+partition and there are messages being fetched about other partitions (#393).
+
+
 # confluent-kafka-javascript 1.6.0
 
 v1.6.0 is a feature release. It is supported for all usage.
diff --git a/MIGRATION.md b/MIGRATION.md
index 27ae6438..1945ae5d 100644
--- a/MIGRATION.md
+++ b/MIGRATION.md
@@ -303,9 +303,11 @@ producerRun().then(consumerRun).catch(console.error);
   - The `heartbeat()` no longer needs to be called by the user in the `eachMessage/eachBatch` callback.
     Heartbeats are automatically managed by librdkafka.
   - The `partitionsConsumedConcurrently` is supported by both `eachMessage` and `eachBatch`.
-  - An API compatible version of `eachBatch` is available, but the batch size calculation is not
-    as per configured parameters, rather, a constant maximum size is configured internally. This is subject
-    to change.
+  - An API compatible version of `eachBatch` is available, maximum batch size
+    can be configured through the `js.consumer.max.batch.size` configuration property
+    and defaults to 32. `js.consumer.max.cache.size.per.worker.ms` allows to
+    configure the cache size estimated based on consumption rate and defaults
+    to 1.5 seconds.
     The property `eachBatchAutoResolve` is supported.
     Within the `eachBatch` callback, use of `uncommittedOffsets` is unsupported,
     and within the returned batch, `offsetLag` and `offsetLagLow` are supported.
diff --git a/ci/update-version.js b/ci/update-version.js
index 531ed2ef..796261f9 100644
--- a/ci/update-version.js
+++ b/ci/update-version.js
@@ -89,7 +89,7 @@ function getPackageVersion(tag, branch) {
 
   // publish with a -devel suffix for EA and RC releases.
   if (tag.prerelease.length > 0) {
-    baseVersion += '-' + tag.prerelease.join('-');
+    baseVersion += '-' + tag.prerelease.join('.');
   }
 
   console.log(`Package version is "${baseVersion}"`);
diff --git a/lib/kafkajs/_consumer.js b/lib/kafkajs/_consumer.js
index ada184f8..d9661ffb 100644
--- a/lib/kafkajs/_consumer.js
+++ b/lib/kafkajs/_consumer.js
@@ -161,11 +161,6 @@ class Consumer {
    */
   #messageCacheMaxSize = 1;
 
-  /**
-   * Number of times we tried to increase the cache.
-   */
-  #increaseCount = 0;
-
   /**
    * Whether the user has enabled manual offset management (commits).
    */
@@ -182,6 +177,20 @@ class Consumer {
    */
   #partitionCount = 0;
 
+  /**
+   * Maximum batch size passed in eachBatch calls.
+   */
+  #maxBatchSize = 32;
+  #maxBatchesSize = 32;
+
+  /**
+   * Maximum cache size in milliseconds per worker.
+   * Based on the consumer rate estimated through the eachMessage/eachBatch calls.
+   *
+   * @default 1500
+   */
+  #maxCacheSizePerWorkerMs = 1500;
+
   /**
    * Whether worker termination has been scheduled.
    */
@@ -234,7 +243,15 @@ class Consumer {
   /**
    * Last fetch real time clock in nanoseconds.
    */
-  #lastFetchClockNs = 0;
+  #lastFetchClockNs = 0n;
+  /**
+   * Last number of messages fetched.
+   */
+  #lastFetchedMessageCnt = 0n;
+  /**
+   * Last fetch concurrency used.
+   */
+  #lastFetchedConcurrency = 0n;
 
   /**
    * List of pending operations to be executed after
@@ -311,8 +328,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) {
@@ -691,6 +706,31 @@ class Consumer {
     this.#cacheExpirationTimeoutMs = this.#maxPollIntervalMs;
     rdKafkaConfig['max.poll.interval.ms'] = this.#maxPollIntervalMs * 2;
 
+    if (rdKafkaConfig['js.consumer.max.batch.size'] !== undefined) {
+      const maxBatchSize = +rdKafkaConfig['js.consumer.max.batch.size'];
+      if (!Number.isInteger(maxBatchSize) || (maxBatchSize <= 0 && maxBatchSize !== -1)) {
+        throw new error.KafkaJSError(
+          "'js.consumer.max.batch.size' must be a positive integer or -1 for unlimited batch size.",
+          { code: error.ErrorCodes.ERR__INVALID_ARG });
+      }
+      this.#maxBatchSize = maxBatchSize;
+      this.#maxBatchesSize = maxBatchSize;
+      if (maxBatchSize === -1) {
+        this.#messageCacheMaxSize = Number.MAX_SAFE_INTEGER;
+      }
+      delete rdKafkaConfig['js.consumer.max.batch.size'];
+    }
+    if (rdKafkaConfig['js.consumer.max.cache.size.per.worker.ms'] !== undefined) {
+      const maxCacheSizePerWorkerMs = +rdKafkaConfig['js.consumer.max.cache.size.per.worker.ms'];
+      if (!Number.isInteger(maxCacheSizePerWorkerMs) || (maxCacheSizePerWorkerMs <= 0)) {
+        throw new error.KafkaJSError(
+          "'js.consumer.max.cache.size.per.worker.ms' must be a positive integer.",
+          { code: error.ErrorCodes.ERR__INVALID_ARG });
+      }
+      this.#maxCacheSizePerWorkerMs = maxCacheSizePerWorkerMs;
+      delete rdKafkaConfig['js.consumer.max.cache.size.per.worker.ms'];
+    }
+
     return rdKafkaConfig;
   }
 
@@ -761,7 +801,7 @@ class Consumer {
    * @returns {import("../../types/kafkajs").EachMessagePayload}
    * @private
    */
-  #createPayload(message) {
+  #createPayload(message, worker) {
     let key = message.key;
     if (typeof key === 'string') {
       key = Buffer.from(key);
@@ -785,6 +825,7 @@ class Consumer {
       },
       heartbeat: async () => { /* no op */ },
       pause: this.pause.bind(this, [{ topic: message.topic, partitions: [message.partition] }]),
+      _worker: worker,
     };
   }
 
@@ -844,40 +885,13 @@ 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.
    * @returns {import("../../types/kafkajs").EachBatchPayload}
    * @private
    */
-  #createBatchPayload(messages) {
+  #createBatchPayload(messages, worker) {
     const topic = messages[0].topic;
     const partition = messages[0].partition;
     let watermarkOffsets = {};
@@ -942,6 +956,7 @@ class Consumer {
       _stale: false,
       _seeked: false,
       _lastResolvedOffset: { offset: -1, leaderEpoch: -1 },
+      _worker: worker,
       heartbeat: async () => { /* no op */ },
       pause: this.pause.bind(this, [{ topic, partitions: [partition] }]),
       commitOffsetsIfNecessary: this.#eachBatchPayload_commitOffsetsIfNecessary.bind(this),
@@ -957,6 +972,43 @@ class Consumer {
     return returnPayload;
   }
 
+  #updateMaxMessageCacheSize() {
+    if (this.#maxBatchSize === -1) {
+      // In case of unbounded max batch size it returns all available messages
+      // for a partition in each batch. Cache is unbounded given that
+      // it takes only one call to process each partition.
+      return;
+    }
+
+    const nowNs = hrtime.bigint();
+    if (this.#lastFetchedMessageCnt > 0 && this.#lastFetchClockNs > 0n &&
+        nowNs > this.#lastFetchClockNs) {
+      const consumptionDurationMilliseconds = Number(nowNs - this.#lastFetchClockNs) / 1e6;
+      const messagesPerMillisecondSingleWorker = this.#lastFetchedMessageCnt / this.#lastFetchedConcurrency / consumptionDurationMilliseconds;
+      // Keep enough messages in the cache for this.#maxCacheSizePerWorkerMs of concurrent consumption by all workers.
+      // Round up to the nearest multiple of `#maxBatchesSize`.
+      this.#messageCacheMaxSize = Math.ceil(
+        Math.round(this.#maxCacheSizePerWorkerMs * messagesPerMillisecondSingleWorker) * this.#concurrency
+        / this.#maxBatchesSize
+      ) * this.#maxBatchesSize;
+    }
+  }
+
+  #saveFetchStats(messages) {
+    this.#lastFetchClockNs = hrtime.bigint();
+    const partitionsNum = new Map();
+    for (const msg of messages) {
+      const key = partitionKey(msg);
+      partitionsNum.set(key, 1);
+      if (partitionsNum.size >= this.#concurrency) {
+        break;
+      }
+    }
+    this.#lastFetchedConcurrency = partitionsNum.size;
+    this.#lastFetchedMessageCnt = messages.length;
+  }
+
+
   async #fetchAndResolveWith(takeFromCache, size) {
     if (this.#fetchInProgress) {
       await this.#fetchInProgress;
@@ -983,6 +1035,8 @@ class Consumer {
       const fetchResult = new DeferredPromise();
       this.#logger.debug(`Attempting to fetch ${size} messages to the message cache`,
         this.#createConsumerBindingMessageMetadata());
+
+      this.#updateMaxMessageCacheSize();
       this.#internalClient.consume(size, (err, messages) =>
         fetchResult.resolve([err, messages]));
 
@@ -999,13 +1053,8 @@ class Consumer {
 
       this.#messageCache.addMessages(messages);
       const res = takeFromCache();
-      this.#lastFetchClockNs = hrtime.bigint();
+      this.#saveFetchStats(messages);
       this.#maxPollIntervalRestart.resolve();
-      if (messages.length === this.#messageCacheMaxSize) {
-        this.#increaseMaxSize();
-      } else {
-        this.#decreaseMaxSize(messages.length);
-      }
       return res;
     } finally {
       this.#fetchInProgress.resolve();
@@ -1074,24 +1123,6 @@ class Consumer {
       this.#messageCacheMaxSize);
   }
 
-  /**
-   * Consumes n messages from the internal consumer.
-   * @returns {Promise<import("../..").Message[]>} A promise that resolves to a list of messages. The size of this list is guaranteed to be less than or equal to n.
-   * @note this method cannot be used in conjunction with #consumeSingleCached.
-   * @private
-   */
-  async #consumeN(n) {
-    return new Promise((resolve, reject) => {
-      this.#internalClient.consume(n, (err, messages) => {
-        if (err) {
-          reject(createKafkaJsErrorFromLibRdKafkaError(err));
-          return;
-        }
-        resolve(messages);
-      });
-    });
-  }
-
   /**
    * Flattens a list of topics with partitions into a list of topic, partition.
    * @param {Array<({topic: string, partitions: Array<number>}|{topic: string, partition: number})>} topics
@@ -1262,12 +1293,12 @@ class Consumer {
    * @returns {Promise<number>} The cache index of the message that was processed.
    * @private
    */
-  async #messageProcessor(m, config) {
+  async #messageProcessor(m, config, worker) {
     let ppc;
     [m, ppc] = m;
     let key = partitionKey(m);
     let eachMessageProcessed = false;
-    const payload = this.#createPayload(m);
+    const payload = this.#createPayload(m, worker);
 
     try {
       this.#lastConsumedOffsets.set(key, m);
@@ -1323,11 +1354,11 @@ class Consumer {
    *                                       the passed batch.
    * @private
    */
-  async #batchProcessor(ms, config) {
+  async #batchProcessor(ms, config, worker) {
     let ppc;
     [ms, ppc] = ms;
     const key = partitionKey(ms[0]);
-    const payload = this.#createBatchPayload(ms);
+    const payload = this.#createBatchPayload(ms, worker);
 
     this.#topicPartitionToBatchPayload.set(key, payload);
 
@@ -1392,21 +1423,21 @@ class Consumer {
     return ppc;
   }
 
-  #discardMessages(ms, ppc) {
-    if (ms) {
-      let m = ms[0];
-      if (m.constructor === Array) {
-        m = m[0];
-      }
-      ppc = ms[1];
-      if (m && !this.#lastConsumedOffsets.has(ppc.key)) {
+  #returnMessages(ms) {
+    let m = ms[0];
+    // ppc could have been change we must return it as well.
+    let ppc = ms[1];
+    const messagesToReturn = m.constructor === Array ? m : [m];
+    const firstMessage = messagesToReturn[0];
+    if (firstMessage && !this.#lastConsumedOffsets.has(ppc.key)) {
         this.#lastConsumedOffsets.set(ppc.key, {
-          topic: m.topic,
-          partition: m.partition,
-          offset: m.offset - 1,
+          topic: firstMessage.topic,
+          partition: firstMessage.partition,
+          offset: firstMessage.offset - 1,
         });
-      }
     }
+
+    this.#messageCache.returnMessages(messagesToReturn);
     return ppc;
   }
 
@@ -1446,6 +1477,7 @@ class Consumer {
    */
   async #worker(config, perMessageProcessor, fetcher) {
     let ppc = null;
+    let workerId = Math.random().toString().slice(2);
     while (!this.#workerTerminationScheduled.resolved) {
       try {
         const ms = await fetcher(ppc);
@@ -1453,11 +1485,15 @@ class Consumer {
           continue;
 
         if (this.#pendingOperations.length) {
-          ppc  = this.#discardMessages(ms, ppc);
-          break;
+          /*
+           * Don't process messages anymore, execute the operations first.
+           * Return the messages to the cache that will be cleared if needed.
+           * `ppc` could have been changed, we must return it as well.
+           */
+          ppc  = this.#returnMessages(ms);
+        } else {
+          ppc = await perMessageProcessor(ms, config, workerId);
         }
-
-        ppc = await perMessageProcessor(ms, config);
       } catch (e) {
         /* Since this error cannot be exposed to the user in the current situation, just log and retry.
           * This is due to restartOnFailure being set to always true. */
@@ -1538,7 +1574,9 @@ class Consumer {
    * @private
    */
   async #executePendingOperations() {
-    for (const op of this.#pendingOperations) {
+    // Execute all pending operations, they could add more operations.
+    while (this.#pendingOperations.length > 0) {
+      const op = this.#pendingOperations.shift();
       await op();
     }
     this.#pendingOperations = [];
@@ -1557,16 +1595,18 @@ class Consumer {
    * @private
    */
   async #runInternal(config) {
-    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 fetcher = config.eachMessage
       ? (savedIdx) => this.#consumeSingleCached(savedIdx)
-      : (savedIdx) => this.#consumeCachedN(savedIdx, maxBatchSize);
-    this.#workers = [];
+      : (savedIdx) => this.#consumeCachedN(savedIdx, this.#maxBatchSize);
 
     await this.#lock.write(async () => {
+      this.#workers = [];
+      this.#concurrency = config.partitionsConsumedConcurrently;
+      this.#maxBatchesSize = (
+        config.eachBatch && this.#maxBatchSize > 0 ?
+        this.#maxBatchSize :
+        1) * this.#concurrency;
 
       while (!this.#disconnectStarted) {
         if (this.#maxPollIntervalRestart.resolved)
@@ -1574,6 +1614,8 @@ class Consumer {
 
         this.#workerTerminationScheduled = new DeferredPromise();
         this.#lastFetchClockNs = hrtime.bigint();
+        this.#lastFetchedMessageCnt = 0;
+        this.#lastFetchedConcurrency = 0;
         if (this.#pendingOperations.length === 0) {
           const workersToSpawn = Math.max(1, Math.min(this.#concurrency, this.#partitionCount));
           const cacheExpirationLoop = this.#cacheExpirationLoop();
@@ -1601,38 +1643,6 @@ class Consumer {
     this.#maxPollIntervalRestart.resolve();
   }
 
-  /**
-   * Consumes a single message from the consumer within the given timeout.
-   * THIS METHOD IS NOT IMPLEMENTED.
-   * @note This method cannot be used with run(). Either that, or this must be used.
-   *
-   * @param {any} args
-   * @param {number} args.timeout - the timeout in milliseconds, defaults to 1000.
-   * @returns {import("../..").Message|null} a message, or null if the timeout was reached.
-   * @private
-   */
-  async consume({ timeout } = { timeout: 1000 }) {
-    if (this.#state !== ConsumerState.CONNECTED) {
-      throw new error.KafkaJSError('consume can only be called while connected.', { code: error.ErrorCodes.ERR__STATE });
-    }
-
-    if (this.#running) {
-      throw new error.KafkaJSError('consume() and run() cannot be used together.', { code: error.ErrorCodes.ERR__CONFLICT });
-    }
-
-    this.#internalClient.setDefaultConsumeTimeout(timeout);
-    let m = null;
-
-    try {
-      const ms = await this.#consumeN(1);
-      m = ms[0];
-    } finally {
-      this.#internalClient.setDefaultConsumeTimeout(undefined);
-    }
-
-    throw new error.KafkaJSError('consume() is not implemented.' + m, { code: error.ErrorCodes.ERR__NOT_IMPLEMENTED });
-  }
-
   async #commitOffsetsUntilNoStateErr(offsetsToCommit) {
     let err = { code: error.ErrorCodes.ERR_NO_ERROR };
     do {
diff --git a/lib/kafkajs/_consumer_cache.js b/lib/kafkajs/_consumer_cache.js
index 9047688e..33a2e81c 100644
--- a/lib/kafkajs/_consumer_cache.js
+++ b/lib/kafkajs/_consumer_cache.js
@@ -28,12 +28,19 @@ class PerPartitionMessageCache {
     }
 
     /**
-     * Adds a message to the cache.
+     * Adds a message to the cache as last one.
      */
-    _add(message) {
+    _addLast(message) {
         this.#cache.addLast(message);
     }
 
+    /**
+     * Adds a message to the cache as first one.
+     */
+    _addFirst(message) {
+        this.#cache.addFirst(message);
+    }
+
     get key() {
         return this.#key;
     }
@@ -126,7 +133,7 @@ class MessageCache {
      *
      * @param {Object} message - the message to add to the cache.
      */
-    #add(message) {
+    #add(message, head = false) {
         const key = partitionKey(message);
         let cache = this.#tpToPpc.get(key);
         if (!cache) {
@@ -135,7 +142,11 @@ class MessageCache {
             cache._node = this.#availablePartitions.addLast(cache);
             this.notifyAvailablePartitions();
         }
-        cache._add(message);
+        if (head) {
+            cache._addFirst(message);
+        } else {
+            cache._addLast(message);
+        }
     }
 
     get availableSize() {
@@ -183,7 +194,21 @@ class MessageCache {
      */
     addMessages(messages) {
         for (const message of messages)
-            this.#add(message);
+            this.#add(message, false);
+        this.#size += messages.length;
+    }
+
+    /**
+     * Return messages to the cache, to be read again.
+     *
+     * @param {Array} messages - the messages to return to the cache.
+     */
+    returnMessages(messages) {
+        let i = messages.length - 1;
+        while (i >= 0) {
+            this.#add(messages[i], true);
+            i--;
+        }
         this.#size += messages.length;
     }
 
diff --git a/lib/util.js b/lib/util.js
index ae539b46..d2dcc9e5 100644
--- a/lib/util.js
+++ b/lib/util.js
@@ -52,4 +52,4 @@ util.dictToStringList = function (mapOrObject) {
   return list;
 };
 
-util.bindingVersion = '1.6.0';
+util.bindingVersion = '1.6.1';
diff --git a/package-lock.json b/package-lock.json
index 45f6f0a3..fe9eb280 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -1,12 +1,12 @@
 {
   "name": "@confluentinc/kafka-javascript",
-  "version": "1.6.0",
+  "version": "1.6.1",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "@confluentinc/kafka-javascript",
-      "version": "1.6.0",
+      "version": "1.6.1",
       "hasInstallScript": true,
       "license": "MIT",
       "workspaces": [
@@ -10077,7 +10077,7 @@
     },
     "schemaregistry": {
       "name": "@confluentinc/schemaregistry",
-      "version": "1.6.0",
+      "version": "1.6.1",
       "license": "MIT",
       "dependencies": {
         "@aws-sdk/client-kms": "^3.637.0",
diff --git a/package.json b/package.json
index 4d36d2da..bebafa9c 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@confluentinc/kafka-javascript",
-  "version": "1.6.0",
+  "version": "1.6.1",
   "description": "Node.js bindings for librdkafka",
   "librdkafka": "2.12.0",
   "librdkafka_win": "2.12.0",
diff --git a/schemaregistry/package.json b/schemaregistry/package.json
index fe058a56..b15a5c8f 100644
--- a/schemaregistry/package.json
+++ b/schemaregistry/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@confluentinc/schemaregistry",
-  "version": "1.6.0",
+  "version": "1.6.1",
   "description": "Node.js client for Confluent Schema Registry",
   "main": "dist/index.js",
   "types": "dist/index.d.ts",
diff --git a/test/promisified/admin/fetch_offsets.spec.js b/test/promisified/admin/fetch_offsets.spec.js
index 08e5a81a..c6412530 100644
--- a/test/promisified/admin/fetch_offsets.spec.js
+++ b/test/promisified/admin/fetch_offsets.spec.js
@@ -132,8 +132,7 @@ describe("fetchOffset function", () => {
     await consumer.run({
       eachMessage: async ({ topic, partition, message }) => {
 
-        messagesConsumed.push(message); // Populate messagesConsumed
-        if (messagesConsumed.length === 5) {
+        if (messagesConsumed.length === 4) {
           await consumer.commitOffsets([
             {
               topic,
@@ -142,6 +141,7 @@ describe("fetchOffset function", () => {
             },
           ]);
         }
+        messagesConsumed.push(message); // Populate messagesConsumed
       },
     });
 
diff --git a/test/promisified/consumer/consumeMessages.spec.js b/test/promisified/consumer/consumeMessages.spec.js
index ca9204c6..8ef27e04 100644
--- a/test/promisified/consumer/consumeMessages.spec.js
+++ b/test/promisified/consumer/consumeMessages.spec.js
@@ -412,6 +412,11 @@ describe.each(cases)('Consumer - partitionsConsumedConcurrently = %s -', (partit
             partitions: partitions,
         });
 
+        // If you have a large consume time and consuming one message at a time,
+        // you need to have very small batch sizes to keep the concurrency up.
+        // It's to avoid having a too large cache and postponing the next fetch
+        // and so the rebalance too much.
+        const producer = createProducer({}, {'batch.num.messages': '1'});
         await producer.connect();
         await consumer.connect();
         await consumer.subscribe({ topic: topicName });
@@ -429,9 +434,8 @@ describe.each(cases)('Consumer - partitionsConsumedConcurrently = %s -', (partit
                 inProgressMaxValue = Math.max(inProgress, inProgressMaxValue);
                 if (inProgressMaxValue >= expectedMaxConcurrentWorkers) {
                     maxConcurrentWorkersReached.resolve();
-                } else if (messagesConsumed.length > 2048) {
-                    await sleep(1000);
                 }
+                await sleep(100);
                 inProgress--;
             },
         });
@@ -448,6 +452,7 @@ describe.each(cases)('Consumer - partitionsConsumedConcurrently = %s -', (partit
         await producer.send({ topic: topicName, messages });
         await maxConcurrentWorkersReached;
         expect(inProgressMaxValue).toBe(expectedMaxConcurrentWorkers);
+        await producer.disconnect();
     });
 
     it('consume GZIP messages', async () => {
@@ -612,6 +617,7 @@ describe.each(cases)('Consumer - partitionsConsumedConcurrently = %s -', (partit
         let assigns = 0;
         let revokes = 0;
         let lost = 0;
+        let firstBatchProcessing;
         consumer = createConsumer({
             groupId,
             maxWaitTimeInMs: 100,
@@ -642,9 +648,6 @@ describe.each(cases)('Consumer - partitionsConsumedConcurrently = %s -', (partit
 
         let errors = false;
         let receivedMessages = 0;
-        const batchLengths = [1, 1, 2,
-                              /* cache reset */
-                              1, 1];
         consumer.run({
             partitionsConsumedConcurrently,
             eachBatchAutoResolve: true,
@@ -652,17 +655,14 @@ describe.each(cases)('Consumer - partitionsConsumedConcurrently = %s -', (partit
                 receivedMessages++;
 
                 try {
-                    expect(event.batch.messages.length)
-                        .toEqual(batchLengths[receivedMessages - 1]);
-
-                    if (receivedMessages === 3) {
-                        expect(event.isStale()).toEqual(false);
-                        await sleep(7500);
-                        /* 7.5s 'processing'
-                         * doesn't exceed max poll interval.
-                         * Cache reset is transparent */
-                        expect(event.isStale()).toEqual(false);
-                    }
+                    expect(event.isStale()).toEqual(false);
+                    await sleep(7500);
+                    /* 7.5s 'processing'
+                     * doesn't exceed max poll interval.
+                     * Cache reset is transparent */
+                    expect(event.isStale()).toEqual(false);
+                    if (firstBatchProcessing === undefined)
+                        firstBatchProcessing = receivedMessages;
                 } catch (e) {
                     console.error(e);
                     errors = true;
@@ -686,6 +686,8 @@ describe.each(cases)('Consumer - partitionsConsumedConcurrently = %s -', (partit
         /* Triggers revocation */
         await consumer.disconnect();
 
+        expect(firstBatchProcessing).toBeDefined();
+        expect(receivedMessages).toBeGreaterThan(firstBatchProcessing);
         /* First assignment */
         expect(assigns).toEqual(1);
         /* Revocation on disconnect */
@@ -732,13 +734,7 @@ describe.each(cases)('Consumer - partitionsConsumedConcurrently = %s -', (partit
 
         let errors = false;
         let receivedMessages = 0;
-        const batchLengths = [/* first we reach batches of 32 message and fetches of 64
-                               * max poll interval exceeded happens on second
-                               * 32 messages batch of the 64 msg fetch. */
-                              1, 1, 2, 2, 4, 4, 8, 8, 16, 16, 32, 32, 32, 32,
-                              /* max poll interval exceeded, 32 reprocessed +
-                               * 1 new message. */
-                              1, 1, 2, 2, 4, 4, 8, 8, 3];
+        let firstLongBatchProcessing;
         consumer.run({
             partitionsConsumedConcurrently,
             eachBatchAutoResolve: true,
@@ -746,17 +742,15 @@ describe.each(cases)('Consumer - partitionsConsumedConcurrently = %s -', (partit
                 receivedMessages++;
 
                 try {
-                    expect(event.batch.messages.length)
-                        .toEqual(batchLengths[receivedMessages - 1]);
-
-                    if (receivedMessages === 13) {
+                    if (!firstLongBatchProcessing && event.batch.messages.length >= 32) {
                         expect(event.isStale()).toEqual(false);
                         await sleep(6000);
                         /* 6s 'processing'
                          * cache clearance starts at 7000 */
                         expect(event.isStale()).toEqual(false);
+                        firstLongBatchProcessing = receivedMessages;
                     }
-                    if ( receivedMessages === 14) {
+                    if (firstLongBatchProcessing && receivedMessages === firstLongBatchProcessing + 1) {
                         expect(event.isStale()).toEqual(false);
                         await sleep(10000);
                         /* 10s 'processing'
@@ -791,6 +785,9 @@ describe.each(cases)('Consumer - partitionsConsumedConcurrently = %s -', (partit
         /* Triggers revocation */
         await consumer.disconnect();
 
+        expect(firstLongBatchProcessing).toBeDefined();
+        expect(receivedMessages).toBeGreaterThan(firstLongBatchProcessing);
+
         /* First assignment + assignment after partitions lost */
         expect(assigns).toEqual(2);
         /* Partitions lost + revocation on disconnect */
diff --git a/test/promisified/consumer/consumerCacheTests.spec.js b/test/promisified/consumer/consumerCacheTests.spec.js
index f923c65f..e5bf1950 100644
--- a/test/promisified/consumer/consumerCacheTests.spec.js
+++ b/test/promisified/consumer/consumerCacheTests.spec.js
@@ -142,6 +142,8 @@ describe.each(cases)('Consumer message cache - isAutoCommit = %s - partitionsCon
          * the consumers are created with the same groupId, we create them here.
          * TODO: verify correctness of theory. It's conjecture... which solves flakiness. */
         let groupId = `consumer-group-id-${secureRandom()}`;
+        const multiplier = 18;
+        const numMessages = 16 * multiplier;
         consumer = createConsumer({
             groupId,
             maxWaitTimeInMs: 100,
@@ -164,7 +166,6 @@ describe.each(cases)('Consumer message cache - isAutoCommit = %s - partitionsCon
         const messagesConsumed = [];
         const messagesConsumedConsumer1 = [];
         const messagesConsumedConsumer2 = [];
-        let consumer2ConsumeRunning = false;
 
         consumer.run({
             partitionsConsumedConcurrently,
@@ -176,18 +177,16 @@ describe.each(cases)('Consumer message cache - isAutoCommit = %s - partitionsCon
                         { topic: event.topic, partition: event.partition, offset: Number(event.message.offset) + 1 },
                     ]);
 
-                /* Until the second consumer joins, consume messages slowly so as to not consume them all
-                 * before the rebalance triggers. */
-                if (messagesConsumed.length > 1024 && !consumer2ConsumeRunning) {
-                    await sleep(10);
-                }
+                // Simulate some processing time so we don't poll all messages
+                // and put them in the cache before consumer2 joins.
+                if (messagesConsumedConsumer2.length === 0)
+                    await sleep(100);
             }
         });
 
-        /* Evenly distribute 1024*9 messages across 3 partitions */
+        /* Evenly distribute numMessages messages across 3 partitions */
         let i = 0;
-        const multiplier = 9;
-        const messages = Array(1024 * multiplier)
+        const messages = Array(numMessages)
             .fill()
             .map(() => {
                 const value = secureRandom();
@@ -198,7 +197,7 @@ describe.each(cases)('Consumer message cache - isAutoCommit = %s - partitionsCon
 
         // Wait for the messages - some of them, before starting the
         // second consumer.
-        await waitForMessages(messagesConsumed, { number: 1024 });
+        await waitForMessages(messagesConsumed, { number: 16 });
 
         await consumer2.connect();
         await consumer2.subscribe({ topic: topicName });
@@ -210,18 +209,17 @@ describe.each(cases)('Consumer message cache - isAutoCommit = %s - partitionsCon
         });
 
         await waitFor(() => consumer2.assignment().length > 0, () => null);
-        consumer2ConsumeRunning = true;
 
         /* Now that both consumers have joined, wait for all msgs to be consumed */
-        await waitForMessages(messagesConsumed, { number: 1024 * multiplier });
+        await waitForMessages(messagesConsumed, { number: numMessages });
 
         /* No extra messages should be consumed. */
         await sleep(1000);
-        expect(messagesConsumed.length).toEqual(1024 * multiplier);
+        expect(messagesConsumed.length).toEqual(numMessages);
 
         /* Check if all messages were consumed. */
         expect(messagesConsumed.map(event => (+event.message.offset)).sort((a, b) => a - b))
-            .toEqual(Array(1024 * multiplier).fill().map((_, i) => Math.floor(i / 3)));
+            .toEqual(Array(numMessages).fill().map((_, i) => Math.floor(i / 3)));
 
         /* Consumer2 should have consumed at least one message. */
         expect(messagesConsumedConsumer2.length).toBeGreaterThan(0);
@@ -234,6 +232,10 @@ describe.each(cases)('Consumer message cache - isAutoCommit = %s - partitionsCon
          * non-message events like rebalances, etc. Internally, this is to make sure that
          * we call poll() at least once within max.poll.interval.ms even if the cache is
          * still full. This depends on us expiring the cache on time. */
+
+        /* FIXME: this test can be flaky when using KIP-848 protocol and
+         * auto-commit. To check if there's something to fix about that case.
+         */
         const impatientConsumer = createConsumer({
             groupId,
             maxWaitTimeInMs: 100,
diff --git a/test/promisified/producer/flush.spec.js b/test/promisified/producer/flush.spec.js
index 64a15b9f..e8dec677 100644
--- a/test/promisified/producer/flush.spec.js
+++ b/test/promisified/producer/flush.spec.js
@@ -71,6 +71,10 @@ describe('Producer > Flush', () => {
 
     it('times out if messages are pending',
         async () => {
+           const producer = createProducer({
+            }, {
+                'batch.num.messages': 1,
+            });
             await producer.connect();
             let messageSent = false;
 
@@ -82,6 +86,7 @@ describe('Producer > Flush', () => {
             /* Small timeout */
             await expect(producer.flush({ timeout: 1 })).rejects.toThrow(Kafka.KafkaJSTimeout);
             expect(messageSent).toBe(false);
+            await producer.disconnect();
         }
     );
 
diff --git a/types/kafkajs.d.ts b/types/kafkajs.d.ts
index 6045a77e..c18bbc84 100644
--- a/types/kafkajs.d.ts
+++ b/types/kafkajs.d.ts
@@ -244,7 +244,24 @@ export interface ConsumerConfig {
   partitionAssignors?: PartitionAssignors[],
 }
 
-export type ConsumerGlobalAndTopicConfig = ConsumerGlobalConfig & ConsumerTopicConfig;
+export interface JSConsumerConfig {
+  /**
+   * Maximum batch size passed in eachBatch calls.
+   * A value of -1 means no limit.
+   *
+   * @default 32
+   */
+  'js.consumer.max.batch.size'?: string | number,
+  /**
+   * Maximum cache size per worker in milliseconds based on the
+   * consume rate estimated through the eachMessage/eachBatch calls.
+   *
+   * @default 1500
+   */
+  'js.consumer.max.cache.size.per.worker.ms'?: string | number
+}
+
+export type ConsumerGlobalAndTopicConfig = ConsumerGlobalConfig & ConsumerTopicConfig & JSConsumerConfig;
 
 export interface ConsumerConstructorConfig extends ConsumerGlobalAndTopicConfig {
   kafkaJS?: ConsumerConfig;