diff --git a/CHANGELOG.md b/CHANGELOG.md
index b5823789..19b33882 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,14 @@
+# 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 (#389).
+
+
 # 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/tests/run_perf_test.js b/ci/tests/run_perf_test.js
index b8e9e808..86b78696 100644
--- a/ci/tests/run_perf_test.js
+++ b/ci/tests/run_perf_test.js
@@ -42,8 +42,20 @@ function belowTarget(value, target) {
 async function main() {
   // Run performance tests and store outputs in memory
   const messageCount = process.env.MESSAGE_COUNT ? +process.env.MESSAGE_COUNT : 50000;
-  const skipCtpTest = process.env.SKIP_CTP_TEST ? process.env.SKIP_CTP_TEST === 'true' : false;
+  let skipCtpTest = process.env.SKIP_CTP_TEST ? process.env.SKIP_CTP_TEST === 'true' : false;
   const concurrentRun = process.env.CONCURRENT_RUN ? process.env.CONCURRENT_RUN === 'true' : false;
+  if (concurrentRun) {
+    skipCtpTest = true;
+  }
+  if (!process.env.CONSUMER_MAX_BATCH_SIZE) {
+    process.env.CONSUMER_MAX_BATCH_SIZE = '-1';
+  }
+  if (!process.env.PARTITIONS_CONSUMED_CONCURRENTLY) {
+    process.env.PARTITIONS_CONSUMED_CONCURRENTLY = '2';
+  }
+  if (!process.env.COMPRESSION) {
+    process.env.COMPRESSION = 'GZIP';
+  }
   const consumerMode = process.env.CONSUMER_MODE || 'all';
   const produceToSecondTopic = process.env.PRODUCE_TO_SECOND_TOPIC ? process.env.PRODUCE_TO_SECOND_TOPIC === 'true' : false;
   const produceToSecondTopicParam = produceToSecondTopic ? '--produce-to-second-topic' : '';
@@ -78,24 +90,32 @@ async function main() {
 
     if (concurrentRun) {
       console.log(`Running ${modeLabel} Producer/Consumer test (concurrently)...`);
+      const INITIAL_DELAY_MS = 10000;
       const TERMINATE_TIMEOUT_MS = process.env.TERMINATE_TIMEOUT_MS ? +process.env.TERMINATE_TIMEOUT_MS : 600000;
-      // Wait 2s more to see if all lag is caught up
-      const TERMINATE_TIMEOUT_MS_CONSUMERS = TERMINATE_TIMEOUT_MS + 2000;
+      // Wait INITIAL_DELAY_MS more to see if all lag is caught up, start earlier than the producer to check
+      // E2E latencies more accurately.
+      const TERMINATE_TIMEOUT_MS_CONSUMERS = TERMINATE_TIMEOUT_MS + INITIAL_DELAY_MS + 2000;
+      const TERMINATE_TIMEOUT_MS_LAG_MONITORING = TERMINATE_TIMEOUT_MS + 1000;
 
       await runCommand(`MODE=${mode} node performance-consolidated.js --create-topics`);
+
+      console.log(`Waiting 10s after topic creation before starting producer and consumers...`);
+      await new Promise(resolve => setTimeout(resolve, 10000));
+
+      console.log(`Starting producer and consumers...`);
       const allPromises = [];
-      allPromises.push(runCommand(`MODE=${mode} MESSAGE_COUNT=${messageCount} node performance-consolidated.js --producer`));
+      allPromises.push(runCommand(`MODE=${mode} MESSAGE_COUNT=${messageCount} INITIAL_DELAY_MS=${INITIAL_DELAY_MS} node performance-consolidated.js --producer`));
       if (consumerModeAll || consumerModeEachMessage) {
-        allPromises.push(runCommand(`MODE=${mode} MESSAGE_COUNT=${messageCount} TERMINATE_TIMEOUT_MS=${TERMINATE_TIMEOUT_MS_CONSUMERS} GROUPID_MESSAGE=${groupIdEachMessage} node performance-consolidated.js --consumer-each-message ${produceToSecondTopicParam}`));
+        allPromises.push(runCommand(`MODE=${mode} MESSAGE_COUNT=${messageCount} INITIAL_DELAY_MS=0 TERMINATE_TIMEOUT_MS=${TERMINATE_TIMEOUT_MS_CONSUMERS} GROUPID_MESSAGE=${groupIdEachMessage} node performance-consolidated.js --consumer-each-message ${produceToSecondTopicParam}`));
       }
       if (consumerModeAll || consumerModeEachBatch) {
-        allPromises.push(runCommand(`MODE=${mode} MESSAGE_COUNT=${messageCount} TERMINATE_TIMEOUT_MS=${TERMINATE_TIMEOUT_MS_CONSUMERS} GROUPID_BATCH=${groupIdEachBatch} node performance-consolidated.js --consumer-each-batch ${produceToSecondTopicParam}`));
+        allPromises.push(runCommand(`MODE=${mode} MESSAGE_COUNT=${messageCount} INITIAL_DELAY_MS=0 TERMINATE_TIMEOUT_MS=${TERMINATE_TIMEOUT_MS_CONSUMERS} GROUPID_BATCH=${groupIdEachBatch} node performance-consolidated.js --consumer-each-batch ${produceToSecondTopicParam}`));
       }
       if (consumerModeAll || consumerModeEachMessage) {
-        allPromises.push(runCommand(`MODE=${mode} TERMINATE_TIMEOUT_MS=${TERMINATE_TIMEOUT_MS_CONSUMERS} GROUPID_MONITOR=${groupIdEachMessage} node performance-consolidated.js --monitor-lag`));
+        allPromises.push(runCommand(`MODE=${mode} INITIAL_DELAY_MS=${INITIAL_DELAY_MS} TERMINATE_TIMEOUT_MS=${TERMINATE_TIMEOUT_MS_LAG_MONITORING} GROUPID_MONITOR=${groupIdEachMessage} node performance-consolidated.js --monitor-lag`));
       }
       if (consumerModeAll || consumerModeEachBatch) {
-        allPromises.push(runCommand(`MODE=${mode} TERMINATE_TIMEOUT_MS=${TERMINATE_TIMEOUT_MS_CONSUMERS} GROUPID_MONITOR=${groupIdEachBatch} node performance-consolidated.js --monitor-lag`));
+        allPromises.push(runCommand(`MODE=${mode} INITIAL_DELAY_MS=${INITIAL_DELAY_MS} TERMINATE_TIMEOUT_MS=${TERMINATE_TIMEOUT_MS_LAG_MONITORING} GROUPID_MONITOR=${groupIdEachBatch} node performance-consolidated.js --monitor-lag`));
       }
       const results = await Promise.allSettled(allPromises);
       return results.map(r => r.status === 'fulfilled' ? r.value : '').join('\n');
@@ -119,11 +139,11 @@ async function main() {
 
   console.log('Running Confluent CTP test...');
   const outputConfluentCtp = skipCtpTest ? '' :
-    runCommand('MODE=confluent MESSAGE_COUNT=5000 node performance-consolidated.js --create-topics --ctp');
+    (await runCommand('MODE=confluent MESSAGE_COUNT=5000 node performance-consolidated.js --create-topics --ctp'));
 
   console.log('Running KafkaJS CTP test...');
   const outputKjsCtp = skipCtpTest ? '' :
-    runCommand('MODE=kafkajs MESSAGE_COUNT=5000 node performance-consolidated.js --create-topics --ctp');
+    (await runCommand('MODE=kafkajs MESSAGE_COUNT=5000 node performance-consolidated.js --create-topics --ctp'));
 
   // Extract Confluent results
   let ctpConfluent, ctpKjs;
@@ -232,6 +252,7 @@ async function main() {
     consumerKjsMessageMaxLatencyT0T1 = extractValue(outputKjsProducerConsumer, '=== Consumer max E2E latency T0-T1 (eachMessage):');
     consumerKjsMessageAvgLatencyT0T2 = extractValue(outputKjsProducerConsumer, '=== Consumer average E2E latency T0-T2 (eachMessage):');
     consumerKjsMessageMaxLatencyT0T2 = extractValue(outputKjsProducerConsumer, '=== Consumer max E2E latency T0-T2 (eachMessage):');
+    consumerKjsTime = extractValue(outputKjsProducerConsumer, '=== Consumption time (eachMessage):');
     consumerKjsMessageAverageRSS = extractValue(outputKjsProducerConsumer, '=== Average consumer-each-message RSS KB:');
     consumerKjsMessageMaxRSS = extractValue(outputKjsProducerConsumer, '=== Max consumer-each-message RSS KB:');
     consumerKjsMessageAverageBrokerLag = extractValue(outputKjsProducerConsumer, `=== Average broker lag (${groupIdEachMessageKafkaJS}):`);
@@ -239,7 +260,6 @@ async function main() {
     consumerKjsMessageTotalLagMeasurements = extractValue(outputKjsProducerConsumer, `=== Sample size for broker lag measurement (${groupIdEachMessageKafkaJS}):`);
   }
   if (consumerModeAll || consumerModeEachBatch) {
-    consumerKjsTime = extractValue(outputKjsProducerConsumer, '=== Consumption time (eachMessage):');
     consumerKjsBatch = extractValue(outputKjsProducerConsumer, '=== Consumer Rate MB/s (eachBatch):');
     consumerKjsBatchRate = extractValue(outputKjsProducerConsumer, '=== Consumer Rate msg/s (eachBatch):');
     consumerKjsBatchAvgLatencyT0T1 = extractValue(outputKjsProducerConsumer, '=== Consumer average E2E latency T0-T1 (eachBatch):');
@@ -269,37 +289,41 @@ async function main() {
   if (consumerModeAll || consumerModeEachMessage) {
     console.log(`Consumer rates MB/s (eachMessage): confluent ${consumerConfluentMessage}, kafkajs ${consumerKjsMessage}`);
     console.log(`Consumer rates msg/s (eachMessage): confluent ${consumerConfluentMessageRate}, kafkajs ${consumerKjsMessageRate}`);
-    console.log(`Consumer average E2E latency T0-T1 (eachMessage): confluent ${consumerConfluentMessageAvgLatencyT0T1}, kafkajs ${consumerKjsMessageAvgLatencyT0T1}`);
-    console.log(`Consumer max E2E latency T0-T1 (eachMessage): confluent ${consumerConfluentMessageMaxLatencyT0T1}, kafkajs ${consumerKjsMessageMaxLatencyT0T1}`);
-    if (produceToSecondTopic) {
-      console.log(`Consumer average E2E latency T0-T2 (eachMessage): confluent ${consumerConfluentMessageAvgLatencyT0T2}, kafkajs ${consumerKjsMessageAvgLatencyT0T2}`);
-      console.log(`Consumer max E2E latency T0-T2 (eachMessage): confluent ${consumerConfluentMessageMaxLatencyT0T2}, kafkajs ${consumerKjsMessageMaxLatencyT0T2}`);
-    }
     console.log(`Consumption time (eachMessage): confluent ${consumerConfluentTime}, kafkajs ${consumerKjsTime}`);
     console.log(`Average RSS (eachMessage): confluent ${consumerConfluentMessageAverageRSS}, kafkajs ${consumerKjsMessageAverageRSS}`);
     console.log(`Max RSS (eachMessage): confluent ${consumerConfluentMessageMaxRSS}, kafkajs ${consumerKjsMessageMaxRSS}`);
-    console.log(`Average broker lag (eachMessage): confluent ${consumerConfluentMessageAverageBrokerLag}, kafkajs ${consumerKjsMessageAverageBrokerLag}`);
-    console.log(`Max broker lag (eachMessage): confluent ${consumerConfluentMessageMaxBrokerLag}, kafkajs ${consumerKjsMessageMaxBrokerLag}`);
-    console.log(`Sample size for broker lag measurement (eachMessage): confluent ${consumerConfluentMessageTotalLagMeasurements}, kafkajs ${consumerKjsMessageTotalLagMeasurements}`);
+    if (concurrentRun) {
+      console.log(`Consumer average E2E latency T0-T1 (eachMessage): confluent ${consumerConfluentMessageAvgLatencyT0T1}, kafkajs ${consumerKjsMessageAvgLatencyT0T1}`);
+      console.log(`Consumer max E2E latency T0-T1 (eachMessage): confluent ${consumerConfluentMessageMaxLatencyT0T1}, kafkajs ${consumerKjsMessageMaxLatencyT0T1}`);
+      if (produceToSecondTopic) {
+        console.log(`Consumer average E2E latency T0-T2 (eachMessage): confluent ${consumerConfluentMessageAvgLatencyT0T2}, kafkajs ${consumerKjsMessageAvgLatencyT0T2}`);
+        console.log(`Consumer max E2E latency T0-T2 (eachMessage): confluent ${consumerConfluentMessageMaxLatencyT0T2}, kafkajs ${consumerKjsMessageMaxLatencyT0T2}`);
+      }
+      console.log(`Average broker lag (eachMessage): confluent ${consumerConfluentMessageAverageBrokerLag}, kafkajs ${consumerKjsMessageAverageBrokerLag}`);
+      console.log(`Max broker lag (eachMessage): confluent ${consumerConfluentMessageMaxBrokerLag}, kafkajs ${consumerKjsMessageMaxBrokerLag}`);
+      console.log(`Sample size for broker lag measurement (eachMessage): confluent ${consumerConfluentMessageTotalLagMeasurements}, kafkajs ${consumerKjsMessageTotalLagMeasurements}`);
+    }
   }
   if (consumerModeAll || consumerModeEachBatch) {
     console.log(`Consumer rates MB/s (eachBatch): confluent ${consumerConfluentBatch}, kafkajs ${consumerKjsBatch}`);
     console.log(`Consumer rates msg/s (eachBatch): confluent ${consumerConfluentBatchRate}, kafkajs ${consumerKjsBatchRate}`);
-    console.log(`Consumer average E2E latency T0-T1 (eachBatch): confluent ${consumerConfluentBatchAvgLatencyT0T1}, kafkajs ${consumerKjsBatchAvgLatencyT0T1}`);
-    console.log(`Consumer max E2E latency T0-T1 (eachBatch): confluent ${consumerConfluentBatchMaxLatencyT0T1}, kafkajs ${consumerKjsBatchMaxLatencyT0T1}`);
-    if (produceToSecondTopic) {
-      console.log(`Consumer average E2E latency T0-T2 (eachBatch): confluent ${consumerConfluentBatchAvgLatencyT0T2}, kafkajs ${consumerKjsBatchAvgLatencyT0T2}`);
-      console.log(`Consumer max E2E latency T0-T2 (eachBatch): confluent ${consumerConfluentBatchMaxLatencyT0T2}, kafkajs ${consumerKjsBatchMaxLatencyT0T2}`);
-    }
     console.log(`Consumption time (eachBatch): confluent ${consumerConfluentBatchTime}, kafkajs ${consumerKjsBatchTime}`);
-    console.log(`Average eachBatch lag: confluent ${consumerConfluentBatchAverageLag}, kafkajs ${consumerKjsBatchAverageLag}`);
-    console.log(`Max eachBatch lag: confluent ${consumerConfluentBatchMaxLag}, kafkajs ${consumerKjsBatchMaxLag}`);
     console.log(`Average eachBatch size: confluent ${consumerConfluentBatchAverageSize}, kafkajs ${consumerKjsBatchAverageSize}`);
     console.log(`Average RSS (eachBatch): confluent ${consumerConfluentBatchAverageRSS}, kafkajs ${consumerKjsBatchAverageRSS}`);
     console.log(`Max RSS (eachBatch): confluent ${consumerConfluentBatchMaxRSS}, kafkajs ${consumerKjsBatchMaxRSS}`);
-    console.log(`Average broker lag (eachBatch): confluent ${consumerConfluentBatchAverageBrokerLag}, kafkajs ${consumerKjsBatchAverageBrokerLag}`);
-    console.log(`Max broker lag (eachBatch): confluent ${consumerConfluentBatchMaxBrokerLag}, kafkajs ${consumerKjsBatchMaxBrokerLag}`);
-    console.log(`Sample size for broker lag measurement (eachBatch): confluent ${consumerConfluentBatchTotalLagMeasurements}, kafkajs ${consumerKjsBatchTotalLagMeasurements}`);
+    if (concurrentRun) {
+      console.log(`Consumer average E2E latency T0-T1 (eachBatch): confluent ${consumerConfluentBatchAvgLatencyT0T1}, kafkajs ${consumerKjsBatchAvgLatencyT0T1}`);
+      console.log(`Consumer max E2E latency T0-T1 (eachBatch): confluent ${consumerConfluentBatchMaxLatencyT0T1}, kafkajs ${consumerKjsBatchMaxLatencyT0T1}`);
+      if (produceToSecondTopic) {
+        console.log(`Consumer average E2E latency T0-T2 (eachBatch): confluent ${consumerConfluentBatchAvgLatencyT0T2}, kafkajs ${consumerKjsBatchAvgLatencyT0T2}`);
+        console.log(`Consumer max E2E latency T0-T2 (eachBatch): confluent ${consumerConfluentBatchMaxLatencyT0T2}, kafkajs ${consumerKjsBatchMaxLatencyT0T2}`);
+      }
+      console.log(`Average eachBatch lag: confluent ${consumerConfluentBatchAverageLag}, kafkajs ${consumerKjsBatchAverageLag}`);
+      console.log(`Max eachBatch lag: confluent ${consumerConfluentBatchMaxLag}, kafkajs ${consumerKjsBatchMaxLag}`);
+      console.log(`Average broker lag (eachBatch): confluent ${consumerConfluentBatchAverageBrokerLag}, kafkajs ${consumerKjsBatchAverageBrokerLag}`);
+      console.log(`Max broker lag (eachBatch): confluent ${consumerConfluentBatchMaxBrokerLag}, kafkajs ${consumerKjsBatchMaxBrokerLag}`);
+      console.log(`Sample size for broker lag measurement (eachBatch): confluent ${consumerConfluentBatchTotalLagMeasurements}, kafkajs ${consumerKjsBatchTotalLagMeasurements}`);
+    }
   }
   if (!concurrentRun) {
     console.log(`Average RSS: confluent ${consumerConfluentAverageRSS}, kafkajs ${consumerKjsAverageRSS}`);
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/examples/performance/performance-consolidated.js b/examples/performance/performance-consolidated.js
index 4684a709..53fec870 100644
--- a/examples/performance/performance-consolidated.js
+++ b/examples/performance/performance-consolidated.js
@@ -17,14 +17,15 @@ const brokers = process.env.KAFKA_BROKERS || 'localhost:9092';
 const securityProtocol = process.env.SECURITY_PROTOCOL;
 const saslUsername = process.env.SASL_USERNAME;
 const saslPassword = process.env.SASL_PASSWORD;
-const topic = process.env.KAFKA_TOPIC || 'test-topic';
-const topic2 = process.env.KAFKA_TOPIC2 || 'test-topic2';
+const topic = process.env.KAFKA_TOPIC || `test-topic-${mode}`;
+const topic2 = process.env.KAFKA_TOPIC2 || `test-topic2-${mode}`;
 const messageCount = process.env.MESSAGE_COUNT ? +process.env.MESSAGE_COUNT : 1000000;
-const messageSize = process.env.MESSAGE_SIZE ? +process.env.MESSAGE_SIZE : 256;
-const batchSize = process.env.BATCH_SIZE ? +process.env.BATCH_SIZE : 100;
+const messageSize = process.env.MESSAGE_SIZE ? +process.env.MESSAGE_SIZE : 4096;
+const batchSize = process.env.PRODUCER_BATCH_SIZE ? +process.env.PRODUCER_BATCH_SIZE : 100;
 const compression = process.env.COMPRESSION || 'None';
 // Between 0 and 1, percentage of random bytes in each message
 const randomness = process.env.RANDOMNESS ? +process.env.RANDOMNESS : 0.5;
+const initialDelayMs = process.env.INITIAL_DELAY_MS ? +process.env.INITIAL_DELAY_MS : 0;
 const numPartitions = process.env.PARTITIONS ? +process.env.PARTITIONS : 3;
 const partitionsConsumedConcurrently = process.env.PARTITIONS_CONSUMED_CONCURRENTLY ? +process.env.PARTITIONS_CONSUMED_CONCURRENTLY : 1;
 const warmupMessages = process.env.WARMUP_MESSAGES ? +process.env.WARMUP_MESSAGES : (batchSize * 10);
@@ -52,6 +53,13 @@ function logParameters(parameters) {
     }
 }
 
+function printPercentiles(percentiles, type) {
+    for (const { percentile, value, count, total } of percentiles) {
+        const percentileStr = `P${percentile}`.padStart(6, ' ');
+        console.log(`=== Consumer ${percentileStr} E2E latency ${type}: ${value.toFixed(2)} ms (${count}/${total})`);
+    }
+}
+
 (async function () {
     const producer = process.argv.includes('--producer');
     const consumer = process.argv.includes('--consumer');
@@ -101,6 +109,10 @@ function logParameters(parameters) {
     }
 
     console.log(`=== Starting Performance Tests - Mode ${mode} ===`);
+    if (initialDelayMs > 0) {
+        console.log(`=== Initial delay of ${initialDelayMs}ms before starting tests ===`);
+        await new Promise(resolve => setTimeout(resolve, initialDelayMs));
+    }
 
     if (createTopics || all) {
         console.log("=== Creating Topics (deleting if they exist already):");
@@ -164,10 +176,11 @@ function logParameters(parameters) {
         endTrackingMemory('consumer-each-message', `consumer-memory-message-${mode}.json`);
         console.log("=== Consumer Rate MB/s (eachMessage): ", consumerRate);
         console.log("=== Consumer Rate msg/s (eachMessage): ", stats.messageRate);
-        console.log("=== Consumer average E2E latency T0-T1 (eachMessage): ", stats.avgLatencyT0T1);
+        printPercentiles(stats.percentilesTOT1, 'T0-T1 (eachMessage)');
         console.log("=== Consumer max E2E latency T0-T1 (eachMessage): ", stats.maxLatencyT0T1);
         if (produceToSecondTopic) {
             console.log("=== Consumer average E2E latency T0-T2 (eachMessage): ", stats.avgLatencyT0T2);
+            printPercentiles(stats.percentilesTOT2, 'T0-T2 (eachMessage)');
             console.log("=== Consumer max E2E latency T0-T2 (eachMessage): ", stats.maxLatencyT0T2);
         }
         console.log("=== Consumption time (eachMessage): ", stats.durationSeconds);
@@ -192,9 +205,11 @@ function logParameters(parameters) {
         console.log("=== Max eachBatch lag: ", stats.maxOffsetLag);
         console.log("=== Average eachBatch size: ", stats.averageBatchSize);
         console.log("=== Consumer average E2E latency T0-T1 (eachBatch): ", stats.avgLatencyT0T1);
+        printPercentiles(stats.percentilesTOT1, 'T0-T1 (eachBatch)');
         console.log("=== Consumer max E2E latency T0-T1 (eachBatch): ", stats.maxLatencyT0T1);
         if (produceToSecondTopic) {
             console.log("=== Consumer average E2E latency T0-T2 (eachBatch): ", stats.avgLatencyT0T2);
+            printPercentiles(stats.percentilesTOT2, 'T0-T2 (eachBatch)');
             console.log("=== Consumer max E2E latency T0-T2 (eachBatch): ", stats.maxLatencyT0T2);
         }
         console.log("=== Consumption time (eachBatch): ", stats.durationSeconds);
@@ -207,7 +222,9 @@ function logParameters(parameters) {
         console.log(`  ProduceTopic: ${topic2}`);
         console.log(`  Message Count: ${messageCount}`);
         // Seed the topic with messages
-        await runProducer(parameters, topic, batchSize, warmupMessages, messageCount, messageSize, compression);
+        await runProducerCKJS(parameters, topic, batchSize,
+            warmupMessages, messageCount, messageSize, compression,
+            randomness, limitRPS);
         startTrackingMemory();
         const ctpRate = await runConsumeTransformProduce(parameters, topic, topic2, warmupMessages, messageCount, messageProcessTimeMs, ctpConcurrency);
         endTrackingMemory('consume-transform-produce', `consume-transform-produce-${mode}.json`);
diff --git a/examples/performance/performance-primitives-common.js b/examples/performance/performance-primitives-common.js
index 1103a1fa..bde4d900 100644
--- a/examples/performance/performance-primitives-common.js
+++ b/examples/performance/performance-primitives-common.js
@@ -1,9 +1,12 @@
 const { hrtime } = require('process');
 const { randomBytes } = require('crypto');
+const PERCENTILES = [50, 75, 90, 95, 99, 99.9, 99.99, 100];
 
 const TERMINATE_TIMEOUT_MS = process.env.TERMINATE_TIMEOUT_MS ? +process.env.TERMINATE_TIMEOUT_MS : 600000;
 const AUTO_COMMIT = process.env.AUTO_COMMIT || 'false';
 const AUTO_COMMIT_ON_BATCH_END = process.env.AUTO_COMMIT_ON_BATCH_END === 'true';
+const USE_KEYS = process.env.USE_KEYS === 'true';
+
 let autoCommit;
 if (AUTO_COMMIT && AUTO_COMMIT === 'false')
     autoCommit = null;
@@ -14,7 +17,7 @@ else {
     }
 }
 
-function installHandlers() {
+function installHandlers(useTerminateTimeout) {
     const handlers = {
         terminationRequested: false,
         terminateTimeout: null,
@@ -26,8 +29,10 @@ function installHandlers() {
     process.on('SIGINT', terminationRequestedCallback);
     process.on('SIGTERM', terminationRequestedCallback);
     handlers.terminationRequestedCallback = terminationRequestedCallback;
-    handlers.terminateTimeout = setTimeout(terminationRequestedCallback,
-                                  TERMINATE_TIMEOUT_MS);
+    if (useTerminateTimeout) {
+        handlers.terminateTimeout = setTimeout(terminationRequestedCallback,
+            TERMINATE_TIMEOUT_MS);
+    }
     return handlers;
 }
 
@@ -56,10 +61,102 @@ function genericProduceToTopic(producer, topic, messages) {
     });
 }
 
+
+// We use a simple count-sketch for latency percentiles to avoid storing all latencies in memory.
+// because we're also measuring the memory usage of the consumer as part of the performance tests.
+class LatencyCountSketch {
+    #numBuckets;
+    #minValue;
+    #maxValue;
+    #buckets;
+    #counts;
+    #changeBaseLogarithm;
+    #totalCount = 0;
+    #base;
+
+    constructor({
+        error = 0.01, // 1% error
+        minValue = 0.01, // min 10μs latency
+        maxValue = 60000, // max 60s latency
+    }) {
+        // Each bucket represents [x, x * (1 + error))
+        this.#base = 1 + error;
+        // Change base from natural log to log base this.#base
+        this.#changeBaseLogarithm =  Math.log(this.#base);
+        this.#numBuckets = Math.ceil(Math.log(maxValue / minValue) / Math.log(this.#base));
+        this.#maxValue = maxValue;
+
+        this.#buckets = new Array(this.#numBuckets + 2).fill(0);
+        this.#buckets[this.#numBuckets + 1] = Number.POSITIVE_INFINITY;
+        this.#buckets[this.#numBuckets] = this.#maxValue;
+        this.#buckets[0] = 0;
+        let i = this.#numBuckets - 1;
+        let currentValue = maxValue;
+        while (i >= 1) {
+            let nextMinimum = currentValue / this.#base;
+            this.#buckets[i] = nextMinimum;
+            currentValue = nextMinimum;
+            i--;
+        }
+        this.#minValue = this.#buckets[1];
+        this.#counts = new Array(this.#numBuckets + 2).fill(0);
+    }
+
+    add(latency) {
+        let idx = 0;
+        if (latency > 0)
+            idx = Math.ceil(Math.log(latency / this.#minValue) / this.#changeBaseLogarithm);
+        idx = (idx < 0) ? 0 :
+              (idx > this.#buckets.length - 2) ? (this.#buckets.length - 2) :
+               idx;
+        
+        this.#counts[idx]++;
+        this.#totalCount++;
+    }
+
+    percentiles(percentilesArray) {
+        const percentileCounts = percentilesArray.map(p => Math.ceil(this.#totalCount * p / 100));
+        const percentileResults = new Array(percentilesArray.length);
+        var totalCountSoFar = 0;
+        let j = 0;
+        let sum = 0;
+        for (let i = 0; i < this.#counts.length; i++) {
+            sum += this.#counts[i];
+        }
+        for (let i = 0; i < percentileCounts.length; i++) {
+            while ((totalCountSoFar < percentileCounts[i]) && (j < this.#counts.length - 1)) {
+                totalCountSoFar += this.#counts[j];
+                j++;
+            }
+            const bucketIndex = (j < this.#counts.length - 1) ? j : this.#counts.length - 2;
+            percentileResults[i] = [this.#buckets[bucketIndex], totalCountSoFar, this.#totalCount];
+        }
+        return percentileResults;
+    }
+}
+
 async function runConsumer(consumer, topic, warmupMessages, totalMessageCnt, eachBatch, partitionsConsumedConcurrently, stats, actionOnMessages) {
-    const handlers = installHandlers();
-    await consumer.connect();
-    await consumer.subscribe({ topic });
+    const handlers = installHandlers(totalMessageCnt === -1);
+    if (stats) {
+        stats.percentilesTOT1 = new LatencyCountSketch({});
+        stats.percentilesTOT2 = new LatencyCountSketch({});
+    }
+    while (true) {
+        try {
+            await consumer.connect();
+            break;
+        } catch (e) {
+            console.error(`Error connecting consumer: ${e}`);
+        }
+    }
+    while (true) {
+        try {
+            await consumer.subscribe({ topic });
+            break;
+        } catch (e) {
+            console.error(`Error subscribing consumer: ${e}`);
+        }
+    }
 
     let messagesReceived = 0;
     let messagesMeasured = 0;
@@ -71,15 +168,29 @@ async function runConsumer(consumer, topic, warmupMessages, totalMessageCnt, eac
     let startTime;
     let rate;
     let consumptionStopped = false;
+    let lastMessageReceivedAt;
     const skippedMessages = warmupMessages;
-    const decoder = new TextDecoder('utf-8');
 
     const updateLatency = (receivedAt, numMessages, message, isT0T2) => {
         if (!stats)
             return;
 
-        const sentAt = Number(decoder.decode(message.value.slice(0, 13)));
-        const latency = receivedAt - sentAt;
+        if (!message.headers || !message.headers['timestamp']) {
+            console.log('WARN: message without timestamp header received, cannot measure latency');
+            return;
+        }
+        const sentAt = Number(message.headers['timestamp']);
+        let latency = receivedAt - sentAt;
+
+        if (isNaN(latency)) {
+            console.log(`WARN: NaN latency received message timestamp: ${message.value.slice(0, 13)}`);
+            return;
+        } else if (latency < 0) {
+            console.log(`WARN: negative latency ${latency} sentAt ${sentAt} receivedAt ${receivedAt}`);
+            latency = 0;
+        } else if (latency > 60000) {
+            console.log(`WARN: received large latency ${latency} sentAt ${sentAt} receivedAt ${receivedAt}`);
+        }
 
         if (!isT0T2) {
             if (!stats.maxLatencyT0T1) {
@@ -89,6 +200,7 @@ async function runConsumer(consumer, topic, warmupMessages, totalMessageCnt, eac
                 stats.maxLatencyT0T1 = Math.max(stats.maxLatencyT0T1, latency);
                 stats.avgLatencyT0T1 = ((stats.avgLatencyT0T1 * (numMessages - 1)) + latency) / numMessages;
             }
+            stats.percentilesTOT1.add(latency);
         } else {
             if (!stats.maxLatencyT0T2) {
                 stats.maxLatencyT0T2 = latency;
@@ -97,6 +209,7 @@ async function runConsumer(consumer, topic, warmupMessages, totalMessageCnt, eac
                 stats.maxLatencyT0T2 = Math.max(stats.maxLatencyT0T2, latency);
                 stats.avgLatencyT0T2 = ((stats.avgLatencyT0T2 * (numMessages - 1)) + latency) / numMessages;
             }
+            stats.percentilesTOT2.add(latency);
         }
     };
 
@@ -109,11 +222,16 @@ async function runConsumer(consumer, topic, warmupMessages, totalMessageCnt, eac
         if (consumptionStopped)
             return;
         consumptionStopped = true;
-        let durationNanos = Number(hrtime.bigint() - startTime);
+        const now = lastMessageReceivedAt || hrtime.bigint();
+        let durationNanos = Number(now - startTime);
         durationSeconds = durationNanos / 1e9;
         rate = (totalMessageSize / durationNanos) * 1e9 / (1024 * 1024); /* MB/s */
         console.log(`Recvd ${messagesMeasured} messages in ${durationSeconds} seconds, ${totalMessageSize} bytes; rate is ${rate} MB/s`);
-        consumer.pause([{ topic }]);
+        try {
+            consumer.pause([{ topic }]);
+        } catch (e) {
+            console.error(`Error pausing consumer: ${e}`);
+        }
     }
 
     console.log("Starting consumer.");
@@ -151,6 +269,8 @@ async function runConsumer(consumer, topic, warmupMessages, totalMessageCnt, eac
         consumeMethod = {
             partitionsConsumedConcurrently,
             eachBatch: async ({ batch }) => {
+                if (!batch.messages)
+                    return;
                 const messagesBeforeBatch = messagesReceived;
                 const topic = batch.topic;
                 const partition = batch.partition;
@@ -168,6 +288,7 @@ async function runConsumer(consumer, topic, warmupMessages, totalMessageCnt, eac
                         messages = messages.slice(messages.length - messagesMeasured);
                     }
                     const now = Date.now();
+                    lastMessageReceivedAt = hrtime.bigint();
                     messagesBase = messagesMeasured - messages.length;
                     let i = 1;
                     for (const message of messages) {
@@ -179,18 +300,13 @@ async function runConsumer(consumer, topic, warmupMessages, totalMessageCnt, eac
                     if (!startTime) {
                         startTime = hrtime.bigint();
                     } else if (totalMessageCnt > 0 && messagesMeasured >= totalMessageCnt) {
-                        let durationNanos = Number(hrtime.bigint() - startTime);
-                        durationSeconds = durationNanos / 1e9;
-                        rate = durationNanos === 0 ? Infinity :
-                            (totalMessageSize / durationNanos) * 1e9 / (1024 * 1024); /* MB/s */
-                        console.log(`Recvd ${messagesMeasured} messages in ${durationSeconds} seconds, ${totalMessageSize} bytes; rate is ${rate} MB/s`);
-                        consumer.pause([{ topic }]);
+                        stopConsuming();
                     }
                 }
 
                 if (actionOnMessages) {
                     await actionOnMessages(batch.messages);
-                    if (messagesMeasured > 0) {
+                    if (messagesMeasured > 0 && messages && messages.length > 0) {
                         let i = 1;
                         const now = Date.now();
                         for (const message of messages) {
@@ -237,25 +353,33 @@ async function runConsumer(consumer, topic, warmupMessages, totalMessageCnt, eac
         stats.messageRate = durationSeconds > 0 ? 
                             (messagesMeasured / durationSeconds) : Infinity;
         stats.durationSeconds = durationSeconds;
+        stats.percentilesTOT1 = stats.percentilesTOT1.percentiles(PERCENTILES).map((value, index) => ({
+            percentile: PERCENTILES[index],
+            value: value[0],
+            count: value[1],
+            total: value[2],
+        }));
+        stats.percentilesTOT2 = stats.percentilesTOT2.percentiles(PERCENTILES).map((value, index) => ({
+            percentile: PERCENTILES[index],
+            value: value[0],
+            count: value[1],
+            total: value[2],
+        }));
     }
     removeHandlers(handlers);
     return rate;
 }
 
 async function runProducer(producer, topic, batchSize, warmupMessages, totalMessageCnt, msgSize, compression, randomness, limitRPS) {
-    const handlers = installHandlers();
+    const handlers = installHandlers(totalMessageCnt === -1);
     let totalMessagesSent = 0;
     let totalBytesSent = 0;
 
-    const encoder = new TextEncoder();
     let staticValueLength = Math.floor(msgSize * (1 - randomness));
-    if (staticValueLength < 13)
-        staticValueLength = 13;
-    let staticValueRemainder = staticValueLength - 13;
-    if (staticValueRemainder > 0) {
-        staticValueRemainder = randomBytes(staticValueRemainder);
+    if (staticValueLength > 0) {
+        staticValueBytes = randomBytes(staticValueLength);
     } else {
-        staticValueRemainder = Buffer.alloc(0);
+        staticValueBytes = Buffer.alloc(0);
     }
 
     let messageCnt = totalMessageCnt;
@@ -266,8 +390,11 @@ async function runProducer(producer, topic, batchSize, warmupMessages, totalMess
     for (let i = 0; i < messageCnt; i++) {
         /* Generate a different random value for each message */
         messages[i] = {
-            value: Buffer.concat([staticValueRemainder, randomBytes(msgSize - staticValueLength)]),
+            value: Buffer.concat([staticValueBytes, randomBytes(msgSize - staticValueLength)]),
         };
+        if (USE_KEYS) {
+            messages[i].key = randomBytes(8);
+        }
     }
 
     await producer.connect();
@@ -299,8 +426,12 @@ async function runProducer(producer, topic, batchSize, warmupMessages, totalMess
             const modifiedMessages = [];
             const batchStart = messagesDispatched % messageCnt;
             for (const msg of messages.slice(batchStart, batchStart + batchSize)) {
-                modifiedMessages.push({ 
-                    value: Buffer.concat([encoder.encode(Date.now().toString()), msg.value])
+                modifiedMessages.push({
+                    key: msg.key,
+                    value: msg.value,
+                    headers: {
+                        'timestamp': Date.now().toString(),
+                    }
                 });
             }
             promises.push(producer.send({
@@ -347,7 +478,7 @@ async function runProducer(producer, topic, batchSize, warmupMessages, totalMess
 }
 
 async function runLagMonitoring(admin, topic) {
-    const handlers = installHandlers();
+    const handlers = installHandlers(true);
     let groupId = process.env.GROUPID_MONITOR;
     if (!groupId) {
         throw new Error("GROUPID_MONITOR environment variable not set");
diff --git a/examples/performance/performance-primitives-kafkajs.js b/examples/performance/performance-primitives-kafkajs.js
index 1b302297..951e5f6d 100644
--- a/examples/performance/performance-primitives-kafkajs.js
+++ b/examples/performance/performance-primitives-kafkajs.js
@@ -1,4 +1,4 @@
-const { Kafka, CompressionTypes } = require('kafkajs');
+const { Kafka, CompressionTypes, logLevel } = require('kafkajs');
 const { randomBytes } = require('crypto');
 const { hrtime } = require('process');
 const {
@@ -20,10 +20,13 @@ module.exports = {
     runProducerConsumerTogether,
 };
 
+const IS_HIGHER_LATENCY_CLUSTER = process.env.IS_HIGHER_LATENCY_CLUSTER === 'true';
+
 function baseConfiguration(parameters) {
     let ret = {
         clientId: 'kafka-test-performance',
         brokers: parameters.brokers.split(','),
+        logLevel: logLevel.WARN,
     };
     if (parameters.securityProtocol &&
         parameters.saslUsername &&
@@ -147,6 +150,9 @@ class CompatibleConsumer {
 
 function newCompatibleConsumer(parameters, eachBatch) {
     const kafka = new Kafka(baseConfiguration(parameters));
+    const higherLatencyClusterOpts = IS_HIGHER_LATENCY_CLUSTER ? {
+        maxBytesPerPartition: 8388608
+    } : {};
 
     let groupId = eachBatch ? process.env.GROUPID_BATCH : process.env.GROUPID_MESSAGE;
     if (!groupId) {
@@ -154,6 +160,7 @@ function newCompatibleConsumer(parameters, eachBatch) {
     }
     console.log(`New KafkaJS group id: ${groupId}`);
     const consumer = kafka.consumer({
+        ...higherLatencyClusterOpts,
         groupId,
     });
     return new CompatibleConsumer(consumer);
diff --git a/examples/performance/performance-primitives.js b/examples/performance/performance-primitives.js
index 84d56383..a7349c6a 100644
--- a/examples/performance/performance-primitives.js
+++ b/examples/performance/performance-primitives.js
@@ -19,6 +19,10 @@ module.exports = {
     newCompatibleProducer,
 };
 
+
+const CONSUMER_MAX_BATCH_SIZE = process.env.CONSUMER_MAX_BATCH_SIZE ? +process.env.CONSUMER_MAX_BATCH_SIZE : null;
+const IS_HIGHER_LATENCY_CLUSTER = process.env.IS_HIGHER_LATENCY_CLUSTER === 'true';
+
 function baseConfiguration(parameters) {
     let ret = {
         'client.id': 'kafka-test-performance',
@@ -97,9 +101,16 @@ class CompatibleProducer {
     }
 }
 function newCompatibleProducer(parameters, compression) {
+    const higherLatencyClusterOpts = IS_HIGHER_LATENCY_CLUSTER ? {
+        'linger.ms': '200',
+        'sticky.partitioning.linger.ms': '200',
+        'message.max.bytes': '2148352',
+        'batch.size': '2097152',
+    } : {};
     return new CompatibleProducer(
         new Kafka({
         ...baseConfiguration(parameters),
+        ...higherLatencyClusterOpts,
         'compression.codec': CompressionTypes[compression],
     }).producer());
 }
@@ -146,6 +157,12 @@ function newCompatibleConsumer(parameters, eachBatch) {
     const autoCommitOpts = autoCommit > 0 ? 
         { 'enable.auto.commit': true, 'auto.commit.interval.ms': autoCommit } :
         { 'enable.auto.commit': false };
+    const jsOpts = (eachBatch && CONSUMER_MAX_BATCH_SIZE !== null) ? {
+        'js.consumer.max.batch.size': CONSUMER_MAX_BATCH_SIZE
+    } : {};
+    const higherLatencyClusterOpts = IS_HIGHER_LATENCY_CLUSTER ? {
+        'max.partition.fetch.bytes': '8388608'
+    } : {};
 
     let groupId = eachBatch ? process.env.GROUPID_BATCH : process.env.GROUPID_MESSAGE;
     if (!groupId) {
@@ -157,6 +174,8 @@ function newCompatibleConsumer(parameters, eachBatch) {
         'auto.offset.reset': 'earliest',
         'fetch.queue.backoff.ms': '100',
         ...autoCommitOpts,
+        ...jsOpts,
+        ...higherLatencyClusterOpts,
     });
     return new CompatibleConsumer(consumer);
 }
diff --git a/lib/kafkajs/_consumer.js b/lib/kafkajs/_consumer.js
index ada184f8..333b175c 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.
    */
@@ -196,7 +205,6 @@ class Consumer {
    * The number of partitions to consume concurrently as set by the user, or 1.
    */
   #concurrency = 1;
-
   /**
    * Promise that resolves together with last in progress fetch.
    * It's set to null when no fetch is in progress.
@@ -234,7 +242,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 +327,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 +705,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;
   }
 
@@ -844,33 +883,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.
@@ -957,6 +969,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 +1032,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 +1050,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 +1120,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
@@ -1392,21 +1420,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;
   }
 
@@ -1453,11 +1481,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);
         }
-
-        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 +1570,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 +1591,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 +1610,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 +1639,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/kafkajs/_producer.js b/lib/kafkajs/_producer.js
index bf1e058d..d48c6551 100644
--- a/lib/kafkajs/_producer.js
+++ b/lib/kafkajs/_producer.js
@@ -622,27 +622,32 @@ class Producer {
     const msgPromises = [];
     for (let i = 0; i < sendOptions.messages.length; i++) {
       const msg = sendOptions.messages[i];
-
-      if (!Object.hasOwn(msg, "partition") || msg.partition === null) {
-        msg.partition = -1;
+      let key = msg.key;
+      let value = msg.value;
+      let partition = msg.partition;
+      let headers = msg.headers;
+      let timestamp = msg.timestamp;
+
+      if (partition === undefined || msg.partition === null) {
+        partition = -1;
       }
 
-      if (typeof msg.value === 'string') {
-        msg.value = Buffer.from(msg.value);
+      if (typeof value === 'string') {
+        value = Buffer.from(value);
       }
 
-      if (Object.hasOwn(msg, "timestamp") && msg.timestamp) {
-        msg.timestamp = Number(msg.timestamp);
+      if (timestamp) {
+        timestamp = Number(timestamp);
       } else {
-        msg.timestamp = 0;
+        timestamp = 0;
       }
 
-      msg.headers = convertToRdKafkaHeaders(msg.headers);
+      headers = convertToRdKafkaHeaders(headers);
 
       msgPromises.push(new Promise((resolve, reject) => {
         const opaque = { resolve, reject };
         try {
-          this.#internalClient.produce(sendOptions.topic, msg.partition, msg.value, msg.key, msg.timestamp, opaque, msg.headers);
+          this.#internalClient.produce(sendOptions.topic, partition, value, key, timestamp, opaque, headers);
         } catch (err) {
           reject(err);
         }
diff --git a/lib/util.js b/lib/util.js
index ae539b46..389c3411 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-alpha.1';
diff --git a/package-lock.json b/package-lock.json
index cdf72ab0..8c7bcf24 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-alpha.1",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "@confluentinc/kafka-javascript",
-      "version": "1.6.0",
+      "version": "1.6.1-alpha.1",
       "hasInstallScript": true,
       "license": "MIT",
       "workspaces": [
@@ -9969,7 +9969,7 @@
     },
     "schemaregistry": {
       "name": "@confluentinc/schemaregistry",
-      "version": "1.6.0",
+      "version": "1.6.1-alpha.1",
       "license": "MIT",
       "dependencies": {
         "@aws-sdk/client-kms": "^3.637.0",
diff --git a/package.json b/package.json
index 4d36d2da..0e0e27c3 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@confluentinc/kafka-javascript",
-  "version": "1.6.0",
+  "version": "1.6.1-alpha.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 5f9bd0f9..95cd317e 100644
--- a/schemaregistry/package.json
+++ b/schemaregistry/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@confluentinc/schemaregistry",
-  "version": "1.6.0",
+  "version": "1.6.1-alpha.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..f62bdb4e 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);
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;