diff --git a/.gitignore b/.gitignore
index 07f5f6b2..d49064a2 100644
--- a/.gitignore
+++ b/.gitignore
@@ -21,3 +21,5 @@ coverage
 .nyc_output/
 *lcov.info
 **/lcov-report
+examples/performance/*.json
+*.log
diff --git a/.semaphore/semaphore.yml b/.semaphore/semaphore.yml
index 1457b69d..3eece201 100644
--- a/.semaphore/semaphore.yml
+++ b/.semaphore/semaphore.yml
@@ -191,7 +191,7 @@ blocks:
             - export NODE_OPTIONS='--max-old-space-size=1536'
             - cd examples/performance
             - npm install
-            - bash -c '../../ci/tests/run_perf_test.sh'
+            - node '../../ci/tests/run_perf_test.js'
             - rm -rf ./node_modules
 
   - name: "Linux amd64: Release"
diff --git a/CHANGELOG.md b/CHANGELOG.md
index b5823789..39c7cf2c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,17 @@
+# confluent-kafka-javascript 1.7.0
+
+v1.7.0 is a feature 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).
+3. Messages passed to the producer `send` method are kept constant (#394).
+
+
 # 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..40f96f20 100644
--- a/MIGRATION.md
+++ b/MIGRATION.md
@@ -303,9 +303,14 @@ 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.  It is not dependent on the size of the produced batches
+    present on the broker, as these are constructed client-side. Similar to
+    the Java client configuration `max.poll.records`.
+    `js.consumer.max.cache.size.per.worker.ms` allows to
+    configure the cache size estimated based on consumption rate and defaults
+    to the cache being sized to 1.5s worth of messages.
     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
new file mode 100644
index 00000000..93e534a7
--- /dev/null
+++ b/ci/tests/run_perf_test.js
@@ -0,0 +1,407 @@
+#!/usr/bin/env node
+
+const util = require('util');
+const exec = util.promisify(require('child_process').exec);
+
+async function runCommand(command) {
+  try {
+    const output = await exec(command, { encoding: 'utf8', stdio: 'pipe' });
+    return output.stdout;
+  } catch (error) {
+    return (error.stdout || '') + (error.stderr || '') + (error.message || '');
+  }
+}
+
+function extractValue(content, pattern) {
+  try {
+    const lines = content.split('\n');
+    const matchingLine = lines.find(line => line.includes(pattern));
+    if (matchingLine) {
+      const value = matchingLine.split(':')[1]?.trim();
+      return Number(value || '');
+    }
+    return NaN;
+  } catch (error) {
+    return NaN;
+  }
+}
+
+function belowThreshold(value, target, threshold = 0.7) {
+  if (isNaN(value) || isNaN(target))
+    return false;
+  return value < (target * threshold);
+}
+
+function belowTarget(value, target) {
+  return belowThreshold(value, target, 1);
+}
+
+
+
+
+async function main() {
+  // Run performance tests and store outputs in memory
+  const messageCount = process.env.MESSAGE_COUNT ? +process.env.MESSAGE_COUNT : 50000;
+  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' : '';
+  const consumerModeAll = consumerMode === 'all';
+  const consumerModeEachMessage = consumerMode === 'eachMessage';
+  const consumerModeEachBatch = consumerMode === 'eachBatch';
+  let consumerParam = '--consumer';
+  if (consumerModeEachMessage) {
+    consumerParam = '--consumer-each-message';
+  } else if (consumerModeEachBatch) {
+    consumerParam = '--consumer-each-batch';
+  }
+  let outputConfluentProducerConsumer;
+  let outputKjsProducerConsumer;
+  const groupIdEachMessageConfluent = `test-group-confluent-message-` + Math.random();
+  const groupIdEachBatchConfluent = `test-group-confluent-batch-` + Math.random();
+  const groupIdEachMessageKafkaJS = `test-group-kafkajs-message-` + Math.random();
+  const groupIdEachBatchKafkaJS = `test-group-kafkajs-batch-` + Math.random();
+  if (consumerModeAll || consumerModeEachMessage) {
+    console.log(`Confluent eachMessage group id: ${groupIdEachMessageConfluent}`);
+    console.log(`KafkaJS eachMessage group id: ${groupIdEachMessageKafkaJS}`);
+  }
+  if (consumerModeAll || consumerModeEachBatch) {
+    console.log(`Confluent eachBatch group id: ${groupIdEachBatchConfluent}`);
+    console.log(`KafkaJS eachBatch group id: ${groupIdEachBatchKafkaJS}`);
+  }
+
+  const runProducerConsumerMode = async (mode) => {
+    const modeLabel = mode === 'confluent' ? 'Confluent' : 'KafkaJS';
+    const groupIdEachMessage = mode === 'confluent' ? groupIdEachMessageConfluent : groupIdEachMessageKafkaJS;
+    const groupIdEachBatch = mode === 'confluent' ? groupIdEachBatchConfluent : groupIdEachBatchKafkaJS;
+
+    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 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;
+
+      const createTopicResult = await runCommand(`MODE=${mode} node performance-consolidated.js --create-topics`);
+      let ret = [createTopicResult];
+
+      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} INITIAL_DELAY_MS=${INITIAL_DELAY_MS} node performance-consolidated.js --producer`));
+      if (consumerModeAll || consumerModeEachMessage) {
+        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} 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} 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} 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 ret.concat(results.map(r => r.status === 'fulfilled' ? r.value : '')).join('\n');
+    } else {
+      console.log(`Running ${modeLabel} Producer/Consumer test...`);
+      return runCommand(`MODE=${mode} MESSAGE_COUNT=${messageCount} GROUPID_MESSAGE=${groupIdEachMessage} GROUPID_BATCH=${groupIdEachBatch} node performance-consolidated.js --create-topics ${consumerParam} ${produceToSecondTopicParam} --producer`);
+    }
+  }
+
+  const runProducerConsumer = async () => {
+    console.log(`Running Producer/Consumer tests with Confluent Kafka JS at ${new Date().toISOString()}`);
+    outputConfluentProducerConsumer = await runProducerConsumerMode('confluent');
+    console.log(`Running Producer/Consumer tests with KafkaJS at ${new Date().toISOString()}`);
+    outputKjsProducerConsumer = await runProducerConsumerMode('kafkajs');
+    console.log(`Producer/Consumer tests completed at ${new Date().toISOString()}`);
+  }
+
+  await runProducerConsumer();
+  console.log(outputConfluentProducerConsumer);
+  console.log(outputKjsProducerConsumer);
+
+  console.log('Running Confluent CTP test...');
+  const outputConfluentCtp = skipCtpTest ? '' :
+    (await runCommand('MODE=confluent MESSAGE_COUNT=5000 node performance-consolidated.js --create-topics --ctp'));
+
+  console.log('Running KafkaJS CTP test...');
+  const outputKjsCtp = skipCtpTest ? '' :
+    (await runCommand('MODE=kafkajs MESSAGE_COUNT=5000 node performance-consolidated.js --create-topics --ctp'));
+
+  // Extract Confluent results
+  let ctpConfluent, ctpKjs;
+  let consumerConfluentMessage;
+  let consumerConfluentMessageRate;
+  let consumerConfluentMessageAvgLatencyT0T1;
+  let consumerConfluentMessageMaxLatencyT0T1;
+  let consumerConfluentMessageAvgLatencyT0T2;
+  let consumerConfluentMessageMaxLatencyT0T2;
+  let consumerConfluentTime;
+  let consumerConfluentMessageAverageRSS;
+  let consumerConfluentMessageMaxRSS;
+  let consumerConfluentMessageAverageBrokerLag;
+  let consumerConfluentMessageMaxBrokerLag;
+  let consumerConfluentMessageTotalLagMeasurements;
+
+  let consumerConfluentBatch;
+  let consumerConfluentBatchRate;
+  let consumerConfluentBatchAvgLatencyT0T1;
+  let consumerConfluentBatchMaxLatencyT0T1;
+  let consumerConfluentBatchAvgLatencyT0T2;
+  let consumerConfluentBatchMaxLatencyT0T2;
+  let consumerConfluentBatchTime;
+  let consumerConfluentBatchAverageLag;
+  let consumerConfluentBatchMaxLag;
+  let consumerConfluentBatchAverageSize;
+  let consumerConfluentBatchAverageRSS;
+  let consumerConfluentBatchMaxRSS;
+  let consumerConfluentBatchAverageBrokerLag;
+  let consumerConfluentBatchMaxBrokerLag;
+  let consumerConfluentBatchTotalLagMeasurements;
+
+  const producerConfluent = extractValue(outputConfluentProducerConsumer, '=== Producer Rate:');
+  const producerConfluentAverageRSS = extractValue(outputConfluentProducerConsumer, '=== Average producer RSS KB:');
+  const producerConfluentMaxRSS = extractValue(outputConfluentProducerConsumer, '=== Max producer RSS KB:');
+  if (consumerModeAll || consumerModeEachMessage) {
+    consumerConfluentMessage = extractValue(outputConfluentProducerConsumer, '=== Consumer Rate MB/s (eachMessage):');
+    consumerConfluentMessageRate = extractValue(outputConfluentProducerConsumer, '=== Consumer Rate msg/s (eachMessage):');
+    consumerConfluentMessageAvgLatencyT0T1 = extractValue(outputConfluentProducerConsumer, '=== Consumer average E2E latency T0-T1 (eachMessage):');
+    consumerConfluentMessageMaxLatencyT0T1 = extractValue(outputConfluentProducerConsumer, '=== Consumer max E2E latency T0-T1 (eachMessage):');
+    consumerConfluentMessageAvgLatencyT0T2 = extractValue(outputConfluentProducerConsumer, '=== Consumer average E2E latency T0-T2 (eachMessage):');
+    consumerConfluentMessageMaxLatencyT0T2 = extractValue(outputConfluentProducerConsumer, '=== Consumer max E2E latency T0-T2 (eachMessage):');
+    consumerConfluentTime = extractValue(outputConfluentProducerConsumer, '=== Consumption time (eachMessage):');
+    consumerConfluentMessageAverageRSS = extractValue(outputConfluentProducerConsumer, '=== Average consumer-each-message RSS KB:');
+    consumerConfluentMessageMaxRSS = extractValue(outputConfluentProducerConsumer, '=== Max consumer-each-message RSS KB:');
+    consumerConfluentMessageAverageBrokerLag = extractValue(outputConfluentProducerConsumer, `=== Average broker lag (${groupIdEachMessageConfluent}):`);
+    consumerConfluentMessageMaxBrokerLag = extractValue(outputConfluentProducerConsumer, `=== Max broker lag (${groupIdEachMessageConfluent}):`);
+    consumerConfluentMessageTotalLagMeasurements = extractValue(outputConfluentProducerConsumer, `=== Sample size for broker lag measurement (${groupIdEachMessageConfluent}):`);
+  }
+  if (consumerModeAll || consumerModeEachBatch) {
+    consumerConfluentBatch = extractValue(outputConfluentProducerConsumer, '=== Consumer Rate MB/s (eachBatch):');
+    consumerConfluentBatchRate = extractValue(outputConfluentProducerConsumer, '=== Consumer Rate msg/s (eachBatch):');
+    consumerConfluentBatchAvgLatencyT0T1 = extractValue(outputConfluentProducerConsumer, '=== Consumer average E2E latency T0-T1 (eachBatch):');
+    consumerConfluentBatchMaxLatencyT0T1 = extractValue(outputConfluentProducerConsumer, '=== Consumer max E2E latency T0-T1 (eachBatch):');
+    consumerConfluentBatchAvgLatencyT0T2 = extractValue(outputConfluentProducerConsumer, '=== Consumer average E2E latency T0-T2 (eachBatch):');
+    consumerConfluentBatchMaxLatencyT0T2 = extractValue(outputConfluentProducerConsumer, '=== Consumer max E2E latency T0-T2 (eachBatch):');
+    consumerConfluentBatchTime = extractValue(outputConfluentProducerConsumer, '=== Consumption time (eachBatch):');
+    consumerConfluentBatchAverageLag = extractValue(outputConfluentProducerConsumer, '=== Average eachBatch lag:');
+    consumerConfluentBatchMaxLag = extractValue(outputConfluentProducerConsumer, '=== Max eachBatch lag:');
+    consumerConfluentBatchAverageSize = extractValue(outputConfluentProducerConsumer, '=== Average eachBatch size:');
+    consumerConfluentBatchAverageRSS = extractValue(outputConfluentProducerConsumer, '=== Average consumer-each-batch RSS KB:');
+    consumerConfluentBatchMaxRSS = extractValue(outputConfluentProducerConsumer, '=== Max consumer-each-batch RSS KB:');
+    consumerConfluentBatchAverageBrokerLag = extractValue(outputConfluentProducerConsumer, `=== Average broker lag (${groupIdEachBatchConfluent}):`);
+    consumerConfluentBatchMaxBrokerLag = extractValue(outputConfluentProducerConsumer, `=== Max broker lag (${groupIdEachBatchConfluent}):`);
+    consumerConfluentBatchTotalLagMeasurements = extractValue(outputConfluentProducerConsumer, `=== Sample size for broker lag measurement (${groupIdEachBatchConfluent}):`);
+  }
+  const consumerConfluentAverageRSS = extractValue(outputConfluentProducerConsumer, '=== Max Average RSS across tests:');
+  const consumerConfluentMaxRSS = extractValue(outputConfluentProducerConsumer, '=== Max RSS across tests:');
+  if (!skipCtpTest) {
+    ctpConfluent = extractValue(outputConfluentCtp, '=== Consume-Transform-Produce Rate:');
+  }
+
+  // Extract KafkaJS results
+  let consumerKjsMessage;
+  let consumerKjsMessageRate;
+  let consumerKjsMessageAvgLatencyT0T1;
+  let consumerKjsMessageMaxLatencyT0T1;
+  let consumerKjsTime;
+  let consumerKjsMessageAverageRSS;
+  let consumerKjsMessageMaxRSS;
+  let consumerKjsMessageAverageBrokerLag;
+  let consumerKjsMessageMaxBrokerLag;
+  let consumerKjsMessageTotalLagMeasurements;
+
+  let consumerKjsBatch;
+  let consumerKjsBatchRate;
+  let consumerKjsBatchAvgLatencyT0T1;
+  let consumerKjsBatchMaxLatencyT0T1;
+  let consumerKjsBatchTime;
+  let consumerKjsBatchAverageLag;
+  let consumerKjsBatchMaxLag;
+  let consumerKjsBatchAverageSize;
+  let consumerKjsBatchAverageRSS;
+  let consumerKjsBatchMaxRSS;
+  let consumerKjsBatchAverageBrokerLag;
+  let consumerKjsBatchMaxBrokerLag;
+  let consumerKjsBatchTotalLagMeasurements;
+
+  const producerKjs = extractValue(outputKjsProducerConsumer, '=== Producer Rate:');
+  const producerKjsAverageRSS = extractValue(outputKjsProducerConsumer, '=== Average producer RSS KB:');
+  const producerKjsMaxRSS = extractValue(outputKjsProducerConsumer, '=== Max producer RSS KB:');
+  if (consumerModeAll || consumerModeEachMessage) {
+    consumerKjsMessage = extractValue(outputKjsProducerConsumer, '=== Consumer Rate MB/s (eachMessage):');
+    consumerKjsMessageRate = extractValue(outputKjsProducerConsumer, '=== Consumer Rate msg/s (eachMessage):');
+    consumerKjsMessageAvgLatencyT0T1 = extractValue(outputKjsProducerConsumer, '=== Consumer average E2E latency T0-T1 (eachMessage):');
+    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}):`);
+    consumerKjsMessageMaxBrokerLag = extractValue(outputKjsProducerConsumer, `=== Max broker lag (${groupIdEachMessageKafkaJS}):`);
+    consumerKjsMessageTotalLagMeasurements = extractValue(outputKjsProducerConsumer, `=== Sample size for broker lag measurement (${groupIdEachMessageKafkaJS}):`);
+  }
+  if (consumerModeAll || consumerModeEachBatch) {
+    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):');
+    consumerKjsBatchMaxLatencyT0T1 = extractValue(outputKjsProducerConsumer, '=== Consumer max E2E latency T0-T1 (eachBatch):');
+    consumerKjsBatchAvgLatencyT0T2 = extractValue(outputKjsProducerConsumer, '=== Consumer average E2E latency T0-T2 (eachBatch):');
+    consumerKjsBatchMaxLatencyT0T2 = extractValue(outputKjsProducerConsumer, '=== Consumer max E2E latency T0-T2 (eachBatch):');
+    consumerKjsBatchTime = extractValue(outputKjsProducerConsumer, '=== Consumption time (eachBatch):');
+    consumerKjsBatchAverageLag = extractValue(outputKjsProducerConsumer, '=== Average eachBatch lag:');
+    consumerKjsBatchMaxLag = extractValue(outputKjsProducerConsumer, '=== Max eachBatch lag:');
+    consumerKjsBatchAverageSize = extractValue(outputKjsProducerConsumer, '=== Average eachBatch size:');
+    consumerKjsBatchAverageRSS = extractValue(outputKjsProducerConsumer, '=== Average consumer-each-batch RSS KB:');
+    consumerKjsBatchMaxRSS = extractValue(outputKjsProducerConsumer, '=== Max consumer-each-batch RSS KB:');
+    consumerKjsBatchAverageBrokerLag = extractValue(outputKjsProducerConsumer, `=== Average broker lag (${groupIdEachBatchKafkaJS}):`);
+    consumerKjsBatchMaxBrokerLag = extractValue(outputKjsProducerConsumer, `=== Max broker lag (${groupIdEachBatchKafkaJS}):`);
+    consumerKjsBatchTotalLagMeasurements = extractValue(outputKjsProducerConsumer, `=== Sample size for broker lag measurement (${groupIdEachBatchKafkaJS}):`);
+  }
+  const consumerKjsAverageRSS = extractValue(outputKjsProducerConsumer, '=== Max Average RSS across tests:');
+  const consumerKjsMaxRSS = extractValue(outputKjsProducerConsumer, '=== Max RSS across tests:');
+  if (!skipCtpTest) {
+    ctpKjs = extractValue(outputKjsCtp, '=== Consume-Transform-Produce Rate:');
+  }
+
+  // Print results
+  console.log(`Producer rates: confluent ${producerConfluent}, kafkajs ${producerKjs}`);
+  console.log(`Average RSS (produce): confluent ${producerConfluentAverageRSS}, kafkajs ${producerKjsAverageRSS}`);
+  console.log(`Max RSS (produce): confluent ${producerConfluentMaxRSS}, kafkajs ${producerKjsMaxRSS}`);
+  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(`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}`);
+    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(`Consumption time (eachBatch): confluent ${consumerConfluentBatchTime}, kafkajs ${consumerKjsBatchTime}`);
+    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}`);
+    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}`);
+    console.log(`Max RSS: confluent ${consumerConfluentMaxRSS}, kafkajs ${consumerKjsMaxRSS}`);
+  }
+  if (!skipCtpTest) {
+    console.log(`CTP rates: confluent ${ctpConfluent}, kafkajs ${ctpKjs}`);
+  }
+
+  let errcode = 0;
+  const maxPerformanceDifference = 0.7;
+
+  // Compare against KJS (30% threshold)
+  if (belowThreshold(producerConfluent, producerKjs, maxPerformanceDifference)) {
+    console.log(`Producer rates differ by more than 30%: confluent ${producerConfluent}, kafkajs ${producerKjs}`);
+    errcode = 1;
+  }
+
+  if (consumerModeAll || consumerModeEachMessage) {
+    if (belowThreshold(consumerConfluentMessage, consumerKjsMessage, maxPerformanceDifference)) {
+      console.log(`Consumer rates MB/s (eachMessage) differ by more than 30%: confluent ${consumerConfluentMessage}, kafkajs ${consumerKjsMessage}`);
+      // FIXME: improve consumer performance at least to KafkaJS level
+      errcode = 0;
+    }
+
+    // Lower is better for time
+    if (belowThreshold(consumerKjsTime, consumerConfluentTime, maxPerformanceDifference)) {
+      console.log(`Consumption time (eachMessage) differ by more than 30%: confluent ${consumerConfluentTime}, kafkajs ${consumerKjsTime}`);
+      errcode = 0;
+    }
+  }
+
+  if (consumerModeAll || consumerModeEachBatch) {
+    if (belowThreshold(consumerConfluentBatch, consumerKjsBatch, maxPerformanceDifference)) {
+      console.log(`Consumer rates (eachBatch) differ by more than 30%: confluent ${consumerConfluentBatch}, kafkajs ${consumerKjsBatch}`);
+      errcode = 0;
+    }
+
+    // Lower is better for time
+    if (belowThreshold(consumerKjsBatchTime, consumerConfluentBatchTime, maxPerformanceDifference)) {
+      console.log(`Consumption time (eachBatch) differ by more than 30%: confluent ${consumerConfluentBatchTime}, kafkajs ${consumerKjsBatchTime}`);
+      errcode = 0;
+    }
+  }
+
+  if (!skipCtpTest && belowThreshold(ctpConfluent, ctpKjs, maxPerformanceDifference)) {
+    console.log(`CTP rates differ by more than 30%: confluent ${ctpConfluent}, kafkajs ${ctpKjs}`);
+    errcode = 1;
+  }
+
+  // Compare against target numbers
+  const TARGET_PRODUCE = process.env.TARGET_PRODUCE_PERFORMANCE || '35';
+  const TARGET_CONSUME = process.env.TARGET_CONSUME_PERFORMANCE || '18';
+  const TARGET_CTP = process.env.TARGET_CTP_PERFORMANCE || '0.02';
+
+  if (belowTarget(producerConfluent, TARGET_PRODUCE)) {
+    console.log(`Confluent producer rate is below target: ${producerConfluent}`);
+    errcode = 1;
+  }
+
+  if ((consumerModeAll || consumerModeEachMessage) && belowTarget(consumerConfluentMessage, TARGET_CONSUME)) {
+    console.log(`Confluent consumer rate (eachMessage) is below target: ${consumerConfluentMessage}`);
+    errcode = 1;
+  }
+
+  if ((consumerModeAll || consumerModeEachBatch) && belowTarget(consumerConfluentBatch, TARGET_CONSUME)) {
+    console.log(`Confluent consumer rate (eachBatch) is below target: ${consumerConfluentBatch}`);
+    errcode = 1;
+  }
+
+  if (!skipCtpTest && belowTarget(ctpConfluent, TARGET_CTP)) {
+    console.log(`Confluent CTP rate is below target: ${ctpConfluent}`);
+    errcode = 1;
+  }
+
+  process.exit(errcode);
+}
+
+if (require.main === module)
+  main();
\ No newline at end of file
diff --git a/ci/tests/run_perf_test.sh b/ci/tests/run_perf_test.sh
deleted file mode 100755
index c55a67ab..00000000
--- a/ci/tests/run_perf_test.sh
+++ /dev/null
@@ -1,64 +0,0 @@
-#!/bin/bash
-
-testresultConfluentProducerConsumer=$(mktemp)
-testresultConfluentCtp=$(mktemp)
-testresultKjsProducerConsumer=$(mktemp)
-testresultKjsCtp=$(mktemp)
-
-MODE=confluent MESSAGE_COUNT=500000 node performance-consolidated.js --create-topics --consumer --producer 2>&1 | tee "$testresultConfluentProducerConsumer"
-MODE=kafkajs MESSAGE_COUNT=500000 node performance-consolidated.js --create-topics --consumer --producer 2>&1 | tee "$testresultKjsProducerConsumer"
-MODE=confluent MESSAGE_COUNT=5000 node performance-consolidated.js --create-topics --ctp 2>&1 | tee "$testresultConfluentCtp"
-MODE=kafkajs MESSAGE_COUNT=5000 node performance-consolidated.js --create-topics --ctp 2>&1 | tee "$testresultKjsCtp"
-
-producerConfluent=$(grep "=== Producer Rate:" "$testresultConfluentProducerConsumer" | cut -d':' -f2 | tr -d ' ')
-consumerConfluent=$(grep "=== Consumer Rate:" "$testresultConfluentProducerConsumer" | cut -d':' -f2 | tr -d ' ')
-ctpConfluent=$(grep "=== Consume-Transform-Produce Rate:" "$testresultConfluentCtp" | cut -d':' -f2 | tr -d ' ')
-producerKjs=$(grep "=== Producer Rate:" "$testresultKjsProducerConsumer" | cut -d':' -f2 | tr -d ' ')
-consumerKjs=$(grep "=== Consumer Rate:" "$testresultKjsProducerConsumer" | cut -d':' -f2 | tr -d ' ')
-ctpKjs=$(grep "=== Consume-Transform-Produce Rate:" "$testresultKjsCtp" | cut -d':' -f2 | tr -d ' ')
-
-echo "Producer rates: confluent $producerConfluent, kafkajs $producerKjs"
-echo "Consumer rates: confluent $consumerConfluent, kafkajs $consumerKjs"
-echo "CTP rates: confluent $ctpConfluent, kafkajs $ctpKjs"
-
-errcode=0
-
-# Compare against KJS
-if [[ $(echo "$producerConfluent < $producerKjs * 70 / 100" | bc -l) -eq 1 ]]; then
-  echo "Producer rates differ by more than 30%: confluent $producerConfluent, kafkajs $producerKjs"
-  errcode=1
-fi
-
-if [[ $(echo "$consumerConfluent < $consumerKjs * 70 / 100" | bc -l) -eq 1 ]]; then
-  echo "Consumer rates differ by more than 30%: confluent $consumerConfluent, kafkajs $consumerKjs"
-  # FIXME: improve consumer performance at least to KafkaJS level
-  errcode=0
-fi
-
-if [[ $(echo "$ctpConfluent < $ctpKjs * 70 / 100" | bc -l) -eq 1 ]]; then
-  echo "CTP rates differ by more than 30%: confluent $ctpConfluent, kafkajs $ctpKjs"
-  errcode=1
-fi
-
-# Compare against numbers set within semaphore config
-TARGET_PRODUCE="${TARGET_PRODUCE_PERFORMANCE:-35}"
-TARGET_CONSUME="${TARGET_CONSUME_PERFORMANCE:-18}"
-TARGET_CTP="${TARGET_CTP_PERFORMANCE:-0.02}"
-
-if [[ $(echo "$producerConfluent < $TARGET_PRODUCE" | bc -l) -eq 1 ]]; then
-  echo "Confluent producer rate is below target: $producerConfluent"
-  errcode=1
-fi
-
-if [[ $(echo "$consumerConfluent < $TARGET_CONSUME" | bc -l) -eq 1 ]]; then
-  echo "Confluent consumer rate is below target: $consumerConfluent"
-  errcode=1
-fi
-
-if [[ $(echo "$ctpConfluent < $TARGET_CTP" | bc -l) -eq 1 ]]; then
-  echo "Confluent CTP rate is below target: $ctpConfluent"
-  errcode=1
-fi
-
-exit $errcode
-
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 7ef9bce6..c2191aa6 100644
--- a/examples/performance/performance-consolidated.js
+++ b/examples/performance/performance-consolidated.js
@@ -1,84 +1,248 @@
+const fs = require('fs');
 const mode = process.env.MODE ? process.env.MODE : 'confluent';
 
-let runProducer, runConsumer, runConsumeTransformProduce, runCreateTopics, runProducerConsumerTogether;
+let runProducer, runConsumer, runConsumeTransformProduce, runCreateTopics, runProducerConsumerTogether, runProducerCKJS;
 if (mode === 'confluent') {
-    ({ runProducer, runConsumer, runConsumeTransformProduce, runCreateTopics, runProducerConsumerTogether } = require('./performance-primitives'));
+    ({ runProducer, runConsumer, runConsumeTransformProduce,
+        runCreateTopics, runLagMonitoring,
+        runProducerConsumerTogether } = require('./performance-primitives'));
+    runProducerCKJS = runProducer;
 } else {
-    ({ runProducer, runConsumer, runConsumeTransformProduce, runCreateTopics, runProducerConsumerTogether } = require('./performance-primitives-kafkajs'));
+    ({ runProducer, runConsumer, runConsumeTransformProduce, runProducerConsumerTogether } = require('./performance-primitives-kafkajs'));
+    /* createTopics is more reliable in CKJS */
+    ({ runCreateTopics, runLagMonitoring, runProducer: runProducerCKJS } = require('./performance-primitives'));
 }
 
 const brokers = process.env.KAFKA_BROKERS || 'localhost:9092';
-const topic = process.env.KAFKA_TOPIC || 'test-topic';
-const topic2 = process.env.KAFKA_TOPIC2 || 'test-topic2';
+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-${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';
-const warmupMessages = process.env.WARMUP_MESSAGES ? +process.env.WARMUP_MESSAGES : (batchSize * 10);
+// 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 > 1024) ? 1024 : (batchSize * 10));
 const messageProcessTimeMs = process.env.MESSAGE_PROCESS_TIME_MS ? +process.env.MESSAGE_PROCESS_TIME_MS : 5;
 const ctpConcurrency = process.env.CONSUME_TRANSFORM_PRODUCE_CONCURRENCY ? +process.env.CONSUME_TRANSFORM_PRODUCE_CONCURRENCY : 1;
 const consumerProcessingTime = process.env.CONSUMER_PROCESSING_TIME ? +process.env.CONSUMER_PROCESSING_TIME : 100;
 const producerProcessingTime = process.env.PRODUCER_PROCESSING_TIME ? +process.env.PRODUCER_PROCESSING_TIME : 100;
+const limitRPS = process.env.LIMIT_RPS ? +process.env.LIMIT_RPS : null;
+const useCKJSProducerEverywhere = process.env.USE_CKJS_PRODUCER_EVERYWHERE === 'true';
+const parameters = {
+    brokers,
+    securityProtocol,
+    saslUsername,
+    saslPassword,
+}
+
+function logParameters(parameters) {
+    console.log(`  Brokers: ${parameters.brokers}`);
+    if (parameters.securityProtocol && parameters.saslUsername && parameters.saslPassword) {
+        console.log(`  Security Protocol: ${parameters.securityProtocol}`);
+        console.log(`  SASL Username: ${parameters.saslUsername ? parameters.saslUsername : 'not set'}`);
+        console.log(`  SASL Password: ${parameters.saslPassword ? '******' : 'not set'}`);
+    } else {
+        console.log("  No security protocol configured");
+    }
+}
+
+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');
+    const consumerEachMessage = process.argv.includes('--consumer-each-message');
+    const consumerEachBatch = process.argv.includes('--consumer-each-batch');
+    const produceToSecondTopic = process.argv.includes('--produce-to-second-topic');
     const ctp = process.argv.includes('--ctp');
     const produceConsumeLatency = process.argv.includes('--latency');
     const all = process.argv.includes('--all');
     const createTopics = process.argv.includes('--create-topics');
+    const monitorLag = process.argv.includes('--monitor-lag');
+    let maxAverageRSSKB, maxMaxRSSKB;
+    const stats = {};
+
+    let measures = [];
+    let interval;
+    const startTrackingMemory = () => {
+        interval = setInterval(() => {
+            measures.push({ rss: process.memoryUsage().rss,
+                            timestamp: Date.now() });
+        }, 100);
+    };
+
+    const datapointToJSON = (m) =>
+        ({ rss: m.rss.toString(), timestamp: m.timestamp.toString() });
+
+    const endTrackingMemory = (name, fileName) => {
+        clearInterval(interval);
+        interval = null;
+        const averageRSS = measures.reduce((sum, m) => sum + m.rss, 0) / measures.length;
+        const averageRSSKB = averageRSS / 1024;
+        maxAverageRSSKB = !maxAverageRSSKB ? averageRSSKB : Math.max(averageRSSKB, maxAverageRSSKB);
+        console.log(`=== Average ${name} RSS KB: ${averageRSSKB}`);
+        const max = measures.reduce((prev, current) => (prev.rss > current.rss) ? prev : current);
+        const maxRSSKB = max.rss / 1024;
+        maxMaxRSSKB = !maxMaxRSSKB ? maxRSSKB : Math.max(maxRSSKB, maxMaxRSSKB);
+        console.log(`=== Max ${name} RSS KB: ${maxRSSKB}`);
+        if (fileName) {
+            const measuresJSON = JSON.stringify({
+                measures: measures.map(datapointToJSON),
+                averageRSS: averageRSS.toString(),
+                maxRSS: datapointToJSON(max)
+            }, null, 2);
+            fs.writeFileSync(fileName, measuresJSON);
+        }
+        measures = [];
+    }
+
+    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):");
-        console.log(`  Brokers: ${brokers}`);
+        logParameters(parameters);
         console.log(`  Topic: ${topic}`);
         console.log(`  Topic2: ${topic2}`);
-        await runCreateTopics(brokers, topic, topic2);
+        console.log(`  Partitions: ${numPartitions}`);
+        await runCreateTopics(parameters, topic, topic2, numPartitions);
+    }
+
+    if (monitorLag) {
+        console.log("=== Starting Lag Monitoring:");
+        logParameters(parameters);
+        console.log(`  Topic: ${topic}`);
+        const {
+            averageLag,
+            maxLag,
+            totalMeasurements
+        } = await runLagMonitoring(parameters, topic);
+        const monitoredGroupId = process.env.GROUPID_MONITOR;
+        console.log(`=== Average broker lag (${monitoredGroupId}): `, averageLag);
+        console.log(`=== Max broker lag (${monitoredGroupId}): `, maxLag);
+        console.log(`=== Sample size for broker lag measurement (${monitoredGroupId}): `, totalMeasurements);
     }
 
     if (producer || all) {
         console.log("=== Running Basic Producer Performance Test:")
-        console.log(`  Brokers: ${brokers}`);
+        logParameters(parameters);
         console.log(`  Topic: ${topic}`);
         console.log(`  Message Count: ${messageCount}`);
         console.log(`  Message Size: ${messageSize}`);
         console.log(`  Batch Size: ${batchSize}`);
         console.log(`  Compression: ${compression}`);
+        console.log(`  Limit RPS: ${limitRPS}`);
         console.log(`  Warmup Messages: ${warmupMessages}`);
-        const producerRate = await runProducer(brokers, topic, batchSize, warmupMessages, messageCount, messageSize, compression);
+        console.log(`  Use CKJS Producer Everywhere: ${useCKJSProducerEverywhere}`);
+        startTrackingMemory();
+        let runProducerFunction = runProducer;
+        if (useCKJSProducerEverywhere) {
+            runProducerFunction = runProducerCKJS;
+        }
+        const producerRate = await runProducerFunction(parameters, topic, batchSize,
+            warmupMessages, messageCount, messageSize, compression,
+            randomness, limitRPS);
+        endTrackingMemory('producer', `producer-memory-${mode}.json`);
         console.log("=== Producer Rate: ", producerRate);
     }
 
-    if (consumer || all) {
+    if (consumer || consumerEachMessage || all) {
         // If user runs this without --producer then they are responsible for seeding the topic.
-        console.log("=== Running Basic Consumer Performance Test:")
-        console.log(`  Brokers: ${brokers}`);
+        console.log("=== Running Basic Consumer Performance Test (eachMessage):")
+        logParameters(parameters);
         console.log(`  Topic: ${topic}`);
         console.log(`  Message Count: ${messageCount}`);
-        const consumerRate = await runConsumer(brokers, topic, messageCount);
-        console.log("=== Consumer Rate: ", consumerRate);
+        console.log(`  Partitions consumed concurrently: ${partitionsConsumedConcurrently}`);
+        startTrackingMemory();
+        const consumerRate = await runConsumer(parameters, topic,
+            warmupMessages, messageCount,
+            false, partitionsConsumedConcurrently, stats,
+            produceToSecondTopic ? topic2 : null, compression, useCKJSProducerEverywhere,
+            messageSize, limitRPS);
+        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);
+        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);
+    }
+
+    if (consumer || consumerEachBatch || all) {
+        // If user runs this without --producer then they are responsible for seeding the topic.
+        console.log("=== Running Basic Consumer Performance Test (eachBatch):")
+        logParameters(parameters);
+        console.log(`  Topic: ${topic}`);
+        console.log(`  Message Count: ${messageCount}`);
+        console.log(`  Partitions consumed concurrently: ${partitionsConsumedConcurrently}`);
+        startTrackingMemory();
+        const consumerRate = await runConsumer(parameters, topic,
+            warmupMessages, messageCount,
+            true, partitionsConsumedConcurrently, stats,
+            produceToSecondTopic ? topic2 : null, compression, useCKJSProducerEverywhere,
+            messageSize, limitRPS);
+        endTrackingMemory('consumer-each-batch', `consumer-memory-batch-${mode}.json`);
+        console.log("=== Consumer Rate MB/s (eachBatch): ", consumerRate);
+        console.log("=== Consumer Rate msg/s (eachBatch): ", stats.messageRate);
+        console.log("=== Average eachBatch lag: ", stats.averageOffsetLag);
+        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);
     }
 
     if (ctp || all) {
         console.log("=== Running Consume-Transform-Produce Performance Test:")
-        console.log(`  Brokers: ${brokers}`);
+        logParameters(parameters);
         console.log(`  ConsumeTopic: ${topic}`);
         console.log(`  ProduceTopic: ${topic2}`);
         console.log(`  Message Count: ${messageCount}`);
         // Seed the topic with messages
-        await runProducer(brokers, topic, batchSize, warmupMessages, messageCount, messageSize, compression);
-        const ctpRate = await runConsumeTransformProduce(brokers, topic, topic2, warmupMessages, messageCount, messageProcessTimeMs, ctpConcurrency);
+        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`);
         console.log("=== Consume-Transform-Produce Rate: ", ctpRate);
     }
 
     if (produceConsumeLatency || all) {
         console.log("=== Running Produce-To-Consume Latency Performance Test:")
-        console.log(`  Brokers: ${brokers}`);
+        logParameters(parameters);
         console.log(`  Topic: ${topic}`);
         console.log(`  Message Count: ${messageCount}`);
         console.log(`  Consumer Processing Time: ${consumerProcessingTime}`);
         console.log(`  Producer Processing Time: ${producerProcessingTime}`);
-        const { mean, p50, p90, p95, outliers } = await runProducerConsumerTogether(brokers, topic, messageCount, messageSize, producerProcessingTime, consumerProcessingTime);
+        startTrackingMemory();
+        const { mean, p50, p90, p95, outliers } = await runProducerConsumerTogether(parameters, topic, messageCount, messageSize, producerProcessingTime, consumerProcessingTime);
+        endTrackingMemory('producer-consumer-together', `producer-consumer-together-${mode}.json`);
         console.log(`=== Produce-To-Consume Latency (ms): Mean: ${mean}, P50: ${p50}, P90: ${p90}, P95: ${p95}`);
 
         // The presence of outliers invalidates the mean measurement, and rasies concerns as to why there are any.
@@ -87,4 +251,9 @@ const producerProcessingTime = process.env.PRODUCER_PROCESSING_TIME ? +process.e
             console.log("=== Outliers (ms): ", outliers);
         }
     }
+
+    if (maxAverageRSSKB !== undefined && maxMaxRSSKB !== undefined) {
+        console.log(`=== Max Average RSS across tests: `, maxAverageRSSKB);
+        console.log(`=== Max RSS across tests: `, maxMaxRSSKB);
+    }
 })();
\ No newline at end of file
diff --git a/examples/performance/performance-primitives-common.js b/examples/performance/performance-primitives-common.js
new file mode 100644
index 00000000..cf156aff
--- /dev/null
+++ b/examples/performance/performance-primitives-common.js
@@ -0,0 +1,610 @@
+const { hrtime } = require('process');
+const { randomBytes } = require('crypto');
+const fs = require('fs');
+const { logLevel } = require('../../').KafkaJS;
+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;
+else {
+    autoCommit = Number(AUTO_COMMIT);
+    if (isNaN(autoCommit)) {
+        autoCommit = null;
+    }
+}
+
+class PerformanceLogger {
+  constructor(fileName) {
+    this.logLevel = logLevel.INFO;
+    this.writeStream = fs.createWriteStream(fileName);
+  }
+
+  setLogLevel(logLevel) {
+    this.logLevel = logLevel;
+  }
+
+  info(message, extra) {
+    if (this.logLevel >= logLevel.INFO)
+      this.writeStream.write(`INFO: ${message} ${JSON.stringify(extra)}\n`);
+  }
+
+  error(message, extra) {
+    if (this.logLevel >= logLevel.ERROR)
+      this.writeStream.write(`ERROR: ${message} ${JSON.stringify(extra)}\n`);
+  }
+
+  warn(message, extra) {
+    if (this.logLevel >= logLevel.WARN)
+      this.writeStream.write(`WARN: ${message} ${JSON.stringify(extra)}\n`);
+  }
+
+  debug(message, extra) {
+    if (this.logLevel >= logLevel.DEBUG)
+      this.writeStream.write(`DEBUG: ${message} ${JSON.stringify(extra)}\n`);
+  }
+
+  namespace() {
+    return this;
+  }
+}
+
+function installHandlers(useTerminateTimeout) {
+    const handlers = {
+        terminationRequested: false,
+        terminateTimeout: null,
+    };
+    const terminationRequestedCallback = () => {
+        console.log('Termination requested, waiting for current operation to finish...');
+        handlers.terminationRequested = true;
+    }
+    process.on('SIGINT', terminationRequestedCallback);
+    process.on('SIGTERM', terminationRequestedCallback);
+    handlers.terminationRequestedCallback = terminationRequestedCallback;
+    if (useTerminateTimeout) {
+        handlers.terminateTimeout = setTimeout(terminationRequestedCallback,
+            TERMINATE_TIMEOUT_MS);
+    }
+    return handlers;
+}
+
+function removeHandlers(handlers) {
+    process.off('SIGINT', handlers.terminationRequestedCallback);
+    process.off('SIGTERM', handlers.terminationRequestedCallback);
+    clearTimeout(handlers.terminateTimeout);
+}
+
+function getAutoCommit() {
+    return autoCommit;
+}
+
+function genericProduceToTopic(producer, topic, messages) {
+    return producer.send({
+        topic,
+        messages,
+    }).catch((err) => {
+        if (producer.isQueueFullError(err)) {
+            /* do just send them again */
+            return genericProduceToTopic(producer, topic, messages);
+        } else {
+            console.error(err);
+            throw err;
+        }
+    });
+}
+
+
+// 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(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;
+    let totalMessageSize = 0;
+    let totalBatches = 0;
+    let totalOffsetLag = 0;
+    let maxOffsetLag = 0;
+    let durationSeconds = 0;
+    let startTime;
+    let rate;
+    let consumptionStopped = false;
+    let lastMessageReceivedAt;
+    const skippedMessages = warmupMessages;
+
+    const updateLatency = (receivedAt, numMessages, message, isT0T2) => {
+        if (!stats || numMessages < 1)
+            return;
+
+        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) {
+                stats.maxLatencyT0T1 = latency;
+                stats.avgLatencyT0T1 = latency;
+            } else {
+                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;
+                stats.avgLatencyT0T2 = latency;
+            } else {
+                stats.maxLatencyT0T2 = Math.max(stats.maxLatencyT0T2, latency);
+                stats.avgLatencyT0T2 = ((stats.avgLatencyT0T2 * (numMessages - 1)) + latency) / numMessages;
+            }
+            stats.percentilesTOT2.add(latency);
+        }
+    };
+
+    const shouldTerminate = () => {
+        return handlers.terminationRequested ||
+        (totalMessageCnt > 0 && messagesMeasured >= totalMessageCnt);
+    };
+
+    const stopConsuming = () => {
+        if (consumptionStopped)
+            return;
+        consumptionStopped = true;
+        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`);
+        try {
+            consumer.pause([{ topic }]);
+        } catch (e) {
+            console.error(`Error pausing consumer: ${e}`);
+        }
+    }
+
+    console.log("Starting consumer.");
+    let consumeMethod = {
+        partitionsConsumedConcurrently,
+        eachMessage: async ({ topic, partition, message }) => {
+            messagesReceived++;
+            if (messagesReceived >= skippedMessages) {
+                messagesMeasured++;
+                totalMessageSize += message.value.length;
+                updateLatency(Date.now(), messagesMeasured, message, false);
+
+                if (messagesReceived === skippedMessages) {
+                    startTime = hrtime.bigint();
+                } else if (totalMessageCnt > 0 && messagesMeasured === totalMessageCnt) {
+                    stopConsuming();
+                }
+            }
+
+            if (actionOnMessages) {
+                await actionOnMessages([message]);
+                updateLatency(Date.now(), messagesMeasured, message, true);
+            }
+
+            if (autoCommit !== null && AUTO_COMMIT_ON_BATCH_END) {
+                await consumer.commitOffsetsOnBatchEnd([{
+                    topic,
+                    partition,
+                    offset: (Number(message.offset) + 1).toString(),
+                }]);
+            }
+        }
+    }
+    if (eachBatch) {
+        consumeMethod = {
+            partitionsConsumedConcurrently,
+            eachBatch: async ({ batch }) => {
+                if (!batch.messages)
+                    return;
+                const messagesBeforeBatch = messagesReceived;
+                const topic = batch.topic;
+                const partition = batch.partition;
+                let messagesBase;
+                let messages;
+                messagesReceived += batch.messages.length;
+                if (messagesReceived >= skippedMessages) {
+                    const offsetLag = batch.offsetLag();
+                    totalOffsetLag += Number(offsetLag);
+                    maxOffsetLag = Math.max(offsetLag, maxOffsetLag);
+                    totalBatches++;
+                    messagesMeasured = messagesReceived - skippedMessages;
+                    messages = batch.messages;
+                    if (messagesBeforeBatch < skippedMessages) {
+                        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) {
+                        totalMessageSize += message.value.length;
+                        updateLatency(now, messagesBase + i, message, false);
+                        i++;
+                    }
+
+                    if (!startTime) {
+                        startTime = hrtime.bigint();
+                    } else if (totalMessageCnt > 0 && messagesMeasured >= totalMessageCnt) {
+                        stopConsuming();
+                    }
+                }
+
+                if (actionOnMessages) {
+                    await actionOnMessages(batch.messages);
+                    if (messagesMeasured > 0 && messages && messages.length > 0) {
+                        let i = 1;
+                        const now = Date.now();
+                        for (const message of messages) {
+                            updateLatency(now, messagesBase + i, message, true);
+                            i++;
+                        }
+                    }
+                }
+
+                if (autoCommit !== null && AUTO_COMMIT_ON_BATCH_END) {
+                    await consumer.commitOffsetsOnBatchEnd([{
+                        topic,
+                        partition,
+                        offset: (Number(batch.lastOffset()) + 1).toString(),
+                    }]);
+                }
+            }
+        };
+    }
+
+    consumer.run({
+        ...consumeMethod
+    });
+
+    await new Promise((resolve) => {
+        let interval = setInterval(() => {
+            if (shouldTerminate()) {
+                stopConsuming();
+                clearInterval(interval);
+                resolve();
+            }
+        }, 1000);
+    });
+
+    await consumer.disconnect();
+
+    if (stats) {
+        if (eachBatch) {
+            stats.averageOffsetLag = totalBatches > 0 ? (totalOffsetLag / totalBatches) : 0;
+            stats.maxOffsetLag = maxOffsetLag;
+            stats.averageBatchSize = totalBatches > 0 ? (messagesMeasured / totalBatches) : 0;
+        }
+        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(totalMessageCnt === -1);
+    let totalMessagesSent = 0;
+    let totalBytesSent = 0;
+
+    let staticValueLength = Math.floor(msgSize * (1 - randomness));
+    if (staticValueLength > 0) {
+        staticValueBytes = randomBytes(staticValueLength);
+    } else {
+        staticValueBytes = Buffer.alloc(0);
+    }
+
+    let messageCnt = totalMessageCnt;
+    if (messageCnt === -1) {
+        if (batchSize > 20000)
+            batchSize = 20000;
+        messageCnt = Math.ceil(20000 / batchSize) * batchSize;
+    }
+    const messages = Array(messageCnt);
+    for (let i = 0; i < messageCnt; i++) {
+        /* Generate a different random value for each message */
+        messages[i] = {
+            value: Buffer.concat([staticValueBytes, randomBytes(msgSize - staticValueLength)]),
+        };
+        if (USE_KEYS) {
+            messages[i].key = randomBytes(8);
+        }
+    }
+
+    await producer.connect();
+
+    console.log('Sending ' + warmupMessages + ' warmup messages.');
+    while (warmupMessages > 0) {
+        const toProduce = Math.min(batchSize, warmupMessages);
+        await producer.send({
+            topic,
+            messages: messages.slice(0, toProduce).map((msg) => ({
+                key: msg.key,
+                value: msg.value,
+                headers: {
+                    'timestamp': Date.now().toString(),
+                }
+            }))
+        }, compression);
+        warmupMessages -= toProduce;
+    }
+    console.log('Sent warmup messages');
+
+    // Now that warmup is done, start measuring...
+    let startTime;
+    let promises = [];
+    startTime = hrtime();
+    let messagesDispatched = 0;
+
+    // The double while-loop allows us to send a bunch of messages and then
+    // await them all at once. We need the second while loop to keep sending
+    // in case of queue full errors, which surface only on awaiting.
+    while (totalMessageCnt == -1 || messagesDispatched < totalMessageCnt) {
+        const startTimeBatch = hrtime.bigint();
+        const maxToAwait = limitRPS ? limitRPS : 10000;
+        let messagesNotAwaited = 0;
+        while (totalMessageCnt == -1 || messagesDispatched < totalMessageCnt) {
+            const modifiedMessages = [];
+            const batchStart = messagesDispatched % messageCnt;
+            for (let i = 0; i < batchSize; i++) {
+                const msg = messages[(batchStart + i) % messages.length];
+                modifiedMessages.push({
+                    key: msg.key,
+                    value: msg.value,
+                    headers: {
+                        'timestamp': Date.now().toString(),
+                    }
+                });
+               if (messagesNotAwaited + i + 1 >= maxToAwait)
+                        break;
+            }
+            const toProduce = modifiedMessages.length;
+            promises.push(producer.send({
+                topic,
+                messages: modifiedMessages,
+            }, compression).then(() => {
+                totalMessagesSent += toProduce;
+                totalBytesSent += toProduce * msgSize;
+            }).catch((err) => {
+                if (producer.isQueueFullError(err)) {
+                    /* do nothing, just send them again */
+                    messagesDispatched -= toProduce;
+                } else {
+                    console.error(err);
+                    throw err;
+                }
+            }));
+            messagesDispatched += toProduce;
+            messagesNotAwaited += toProduce;
+            if (handlers.terminationRequested || messagesNotAwaited >= maxToAwait)
+                break;
+        }
+        await Promise.all(promises);
+        promises = [];
+        if (handlers.terminationRequested)
+            break;
+
+        const now = hrtime.bigint();
+        const elapsedBatch = now - startTimeBatch;
+        if (limitRPS && elapsedBatch < 1000000000n) {
+            await new Promise(resolve => setTimeout(resolve, Number(1000000000n - elapsedBatch) / 1e6));
+        }
+    }
+    let elapsed = hrtime(startTime);
+    let durationNanos = elapsed[0] * 1e9 + elapsed[1];
+    let rate = (totalBytesSent / durationNanos) * 1e9 / (1024 * 1024); /* MB/s */
+    console.log(`Sent ${totalMessagesSent} messages, ${totalBytesSent} bytes; rate is ${rate} MB/s`);
+
+    await producer.disconnect();
+    removeHandlers(handlers);
+    return rate;
+}
+
+async function runLagMonitoring(admin, topic) {
+    const handlers = installHandlers(true);
+    let groupId = process.env.GROUPID_MONITOR;
+    if (!groupId) {
+        throw new Error("GROUPID_MONITOR environment variable not set");
+    }
+
+    await admin.connect();
+
+    const fetchTotalLag = async () => {
+        const partitionCompleteLag = {};
+        const partitionHWM = {};
+        const partitionLag = {};
+        let totalLag = 0n;
+        const operations = [
+            admin.fetchTopicOffsets(topic, { timeout: 30000 }),
+            admin.fetchOffsets({ groupId, topics: [topic], timeout: 30000 }),
+        ]
+        let [topicOffsets, groupOffsets] = await Promise.all(operations);
+        groupOffsets = groupOffsets[0];
+
+        for (const partitionOffset of topicOffsets) {
+            partitionHWM[partitionOffset.partition] = BigInt(partitionOffset.high);
+            partitionCompleteLag[partitionOffset.partition] = BigInt(partitionOffset.high) - BigInt(partitionOffset.low);
+        }
+        
+        if (groupOffsets && groupOffsets.partitions) {
+            for (const partitionOffset of groupOffsets.partitions) {
+                const partition = partitionOffset.partition;
+                const hwm = partitionHWM[partition];
+                if (hwm && partitionOffset.offset && hwm >= BigInt(partitionOffset.offset)) {
+                    const currentLag = hwm - BigInt(partitionOffset.offset);
+                    partitionLag[partition] = currentLag;
+                    totalLag += currentLag;
+                }
+            }
+        } else {
+            throw new Error(`No offsets found for group ${groupId} on topic ${topic}`);
+        }
+        for (const partition of Object.keys(partitionHWM)) {
+            if (partitionLag[partition] === undefined) {
+                const currentLag = partitionCompleteLag[partition];
+                partitionLag[partition] = currentLag;
+                totalLag += currentLag;
+            }
+        }
+        return totalLag;
+    }
+
+    let totalAverageLag = 0n;
+    let maxLag = 0n;
+    let totalMeasurements = 0;
+
+    while (!handlers.terminationRequested) {
+        try {
+            const lag = await fetchTotalLag();
+            totalAverageLag += lag;
+            maxLag = lag > maxLag ? lag : maxLag;
+            totalMeasurements++;
+        } catch (e) {
+            console.error(`Error fetching lag: ${e}`);
+        }
+        await new Promise(resolve => setTimeout(resolve, 100));
+    }
+    const averageLag = totalMeasurements > 0 ? (Number(totalAverageLag) / totalMeasurements) : NaN;
+    maxLag = Number(maxLag);
+
+    await admin.disconnect();
+    removeHandlers(handlers);
+    return {
+        averageLag,
+        maxLag,
+        totalMeasurements
+    };
+}
+
+module.exports = {
+    runConsumer,
+    runProducer,
+    runLagMonitoring,
+    genericProduceToTopic,
+    getAutoCommit,
+    PerformanceLogger,
+};
\ No newline at end of file
diff --git a/examples/performance/performance-primitives-kafkajs.js b/examples/performance/performance-primitives-kafkajs.js
index 1dfc530e..b777870d 100644
--- a/examples/performance/performance-primitives-kafkajs.js
+++ b/examples/performance/performance-primitives-kafkajs.js
@@ -1,6 +1,17 @@
-const { Kafka, CompressionTypes } = require('kafkajs');
+const { Kafka, CompressionTypes, logLevel } = require('kafkajs');
 const { randomBytes } = require('crypto');
 const { hrtime } = require('process');
+const {
+    runConsumer: runConsumerCommon,
+    runProducer: runProducerCommon,
+    genericProduceToTopic,
+    getAutoCommit,
+    PerformanceLogger,
+ } = require('./performance-primitives-common');
+
+const {
+  newCompatibleProducer: newCompatibleProducerCKJS,
+} = require('./performance-primitives');
 
 module.exports = {
     runProducer,
@@ -10,21 +21,51 @@ module.exports = {
     runProducerConsumerTogether,
 };
 
-async function runCreateTopics(brokers, topic, topic2) {
-    const kafka = new Kafka({
+const IS_HIGHER_LATENCY_CLUSTER = process.env.IS_HIGHER_LATENCY_CLUSTER === 'true';
+const DEBUG = process.env.DEBUG;
+const ENABLE_LOGGING = DEBUG !== undefined;
+
+function baseConfiguration(parameters) {
+    let ret = {
         clientId: 'kafka-test-performance',
-        brokers: brokers.split(','),
-    });
+        brokers: parameters.brokers.split(','),
+        logLevel: logLevel.WARN,
+    };
+    if (parameters.securityProtocol &&
+        parameters.saslUsername &&
+        parameters.saslPassword) {
+        ret = {
+            ...ret,
+            ssl: parameters.securityProtocol.toLowerCase().includes('ssl'),
+            sasl: {
+                mechanism: 'plain',
+                username: parameters.saslUsername,
+                password: parameters.saslPassword
+            }
+        };
+    }
+    if (parameters.logToFile) {
+        ret.logger = new PerformanceLogger(parameters.logToFile);
+    }
+    return ret;
+}
+
+async function runCreateTopics(parameters, topic, topic2, numPartitions) {
+    const adminParameters = {
+        ...parameters,
+        disableLogging: true,
+    };
+    const kafka = new Kafka(baseConfiguration(adminParameters));
 
     const admin = kafka.admin();
     await admin.connect();
 
     for (let t of [topic, topic2]) {
         let topicCreated = await admin.createTopics({
-            topics: [{ topic: t, numPartitions: 3 }],
+            topics: [{ topic: t, numPartitions }],
         }).catch(console.error);
         if (topicCreated) {
-            console.log(`Created topic ${t}`);
+            console.log(`Created topic ${t} with ${numPartitions} partitions`);
             continue;
         }
 
@@ -33,142 +74,144 @@ async function runCreateTopics(brokers, topic, topic2) {
         await new Promise(resolve => setTimeout(resolve, 1000)); /* Propagate. */
         await admin.createTopics({
             topics: [
-                { topic: t, numPartitions: 3 },
+                { topic: t, numPartitions },
             ],
         }).catch(console.error);
-        console.log(`Created topic ${t}`);
+        console.log(`Created topic ${t} with ${numPartitions} partitions`);
     }
 
     await admin.disconnect();
 }
 
-async function runProducer(brokers, topic, batchSize, warmupMessages, totalMessageCnt, msgSize, compression) {
-    let totalMessagesSent = 0;
-    let totalBytesSent = 0;
-
-    const message = {
-        value: randomBytes(msgSize),
+class CompatibleProducer {
+    constructor(producer) {
+        this.producer = producer;
     }
 
-    const messages = Array(batchSize).fill(message);
+    async connect() {
+        return this.producer.connect();
+    }
 
-    const kafka = new Kafka({
-        clientId: 'kafka-test-performance',
-        brokers: brokers.split(','),
-    });
+    async disconnect() {
+        return this.producer.disconnect();
+    }
 
-    const producer = kafka.producer();
-    await producer.connect();
+    isQueueFullError(err) {
+        return false;
+    }
 
-    console.log('Sending ' + warmupMessages + ' warmup messages.');
-    while (warmupMessages > 0) {
-        await producer.send({
-            topic,
-            messages,
+    send(opts, compression) {
+        return this.producer.send({
+            ...opts,
             compression: CompressionTypes[compression],
         });
-        warmupMessages -= batchSize;
     }
-    console.log('Sent warmup messages');
+}
+function newCompatibleProducer(parameters, compression) {
+    return new CompatibleProducer(
+        new Kafka(baseConfiguration(parameters)
+    ).producer());
+}
 
-    // Now that warmup is done, start measuring...
-    let startTime;
-    let promises = [];
-    startTime = hrtime();
-    let messagesDispatched = 0;
-
-    // The double while-loop allows us to send a bunch of messages and then
-    // await them all at once. We need the second while loop to keep sending
-    // in case of queue full errors, which surface only on awaiting.
-    while (totalMessageCnt == -1 || messagesDispatched < totalMessageCnt) {
-        while (totalMessageCnt == -1 || messagesDispatched < totalMessageCnt) {
-            promises.push(producer.send({
-                topic,
-                messages,
-                compression: CompressionTypes[compression],
-            }).then(() => {
-                totalMessagesSent += batchSize;
-                totalBytesSent += batchSize * msgSize;
-            }).catch((err) => {
-                console.error(err);
-                throw err;
-            }));
-            messagesDispatched += batchSize;
-        }
-        await Promise.all(promises);
+async function runProducer(parameters, topic, batchSize, warmupMessages, totalMessageCnt, msgSize, compression, randomness, limitRPS) {
+    if (ENABLE_LOGGING && !parameters.logToFile) {
+        parameters.logToFile = './kafkajs-producer.log';
     }
-    let elapsed = hrtime(startTime);
-    let durationNanos = elapsed[0] * 1e9 + elapsed[1];
-    let rate = (totalBytesSent / durationNanos) * 1e9 / (1024 * 1024); /* MB/s */
-    console.log(`Sent ${totalMessagesSent} messages, ${totalBytesSent} bytes; rate is ${rate} MB/s`);
-
-    await producer.disconnect();
-    return rate;
+    return runProducerCommon(newCompatibleProducer(parameters, compression), topic, batchSize, warmupMessages, totalMessageCnt, msgSize, compression, randomness, limitRPS);
 }
 
-async function runConsumer(brokers, topic, totalMessageCnt) {
-    const kafka = new Kafka({
-        clientId: 'kafka-test-performance',
-        brokers: brokers.split(','),
-    });
+class CompatibleConsumer {
+    constructor(consumer) {
+        this.consumer = consumer;
+    }
 
-    const consumer = kafka.consumer({
-        groupId: 'test-group' + Math.random(),
-    });
-    await consumer.connect();
-    await consumer.subscribe({ topic, fromBeginning: true });
+    async connect() {
+        return this.consumer.connect();
+    }
 
-    let messagesReceived = 0;
-    let messagesMeasured = 0;
-    let totalMessageSize = 0;
-    let startTime;
-    let rate;
-    const skippedMessages = 100;
+    async disconnect() {
+        return this.consumer.disconnect();
+    }
 
-    console.log("Starting consumer.");
+    async subscribe(opts) {
+        return this.consumer.subscribe({
+            ...opts,
+            fromBeginning: true
+        });
+    }
 
-    consumer.run({
-        autoCommit: false,
-        eachMessage: async ({ topic, partition, message }) => {
-            messagesReceived++;
+    pause(topics) {
+        return this.consumer.pause(topics);
+    }
 
-            if (messagesReceived >= skippedMessages) {
-                messagesMeasured++;
-                totalMessageSize += message.value.length;
+    commitOffsetsOnBatchEnd(offsets) {
+        // do nothing, done by KafkaJS after each* if autoCommit is enabled
+    }
 
-                if (messagesReceived === skippedMessages) {
-                    startTime = hrtime();
-                } else if (messagesMeasured === totalMessageCnt) {
-                    let elapsed = hrtime(startTime);
-                    let durationNanos = elapsed[0] * 1e9 + elapsed[1];
-                    rate = (totalMessageSize / durationNanos) * 1e9 / (1024 * 1024); /* MB/s */
-                    console.log(`Recvd ${messagesMeasured} messages, ${totalMessageSize} bytes; rate is ${rate} MB/s`);
-                    consumer.pause([{ topic }]);
-                }
-            }
-        }
-    });
+    run(opts) {
+        const autoCommit = getAutoCommit();
+        const autoCommitOpts = autoCommit > 0 ? 
+            { autoCommit: true, autoCommitInterval: autoCommit } :
+            { autoCommit: false };
+        return this.consumer.run({
+            ...opts,
+            ...autoCommitOpts,
+        });
+    }
+}
 
-    totalMessageSize = 0;
 
-    await new Promise((resolve) => {
-        let interval = setInterval(() => {
-            if (messagesReceived >= totalMessageCnt) {
-                clearInterval(interval);
-                resolve();
-            }
-        }, 1000);
+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) {
+        groupId = 'test-group' + Math.random();
+    }
+    console.log(`New KafkaJS group id: ${groupId}`);
+    const consumer = kafka.consumer({
+        ...higherLatencyClusterOpts,
+        groupId,
     });
+    return new CompatibleConsumer(consumer);
+}
 
-    await consumer.disconnect();
-    return rate;
+
+async function runConsumer(parameters, topic, warmupMessages, totalMessageCnt, eachBatch, partitionsConsumedConcurrently, stats, produceToTopic, produceCompression, useCKJSProducerEverywhere,
+                           messageSize, limitRPS) {
+    let actionOnMessages = null;
+    let producer;
+    if (produceToTopic) {
+        const newCompatibleProducerFunction = useCKJSProducerEverywhere ?
+            newCompatibleProducerCKJS : newCompatibleProducer;
+        const producerParameters = {
+            ...parameters,
+        };
+        if (ENABLE_LOGGING)
+            producerParameters.logToFile = './kafkajs-consumer-producer.log';
+        producer = newCompatibleProducerFunction(producerParameters, produceCompression);
+        await producer.connect();
+        actionOnMessages = (messages) =>
+            genericProduceToTopic(producer, produceToTopic, messages);
+    }
+    const consumerParameters = {
+        ...parameters,
+    };
+    if (ENABLE_LOGGING)
+        consumerParameters.logToFile = eachBatch ? './kafkajs-consumer-batch.log' :
+                                                   './kafkajs-consumer-message.log';
+    const ret = await runConsumerCommon(newCompatibleConsumer(consumerParameters, eachBatch), topic, warmupMessages, totalMessageCnt, eachBatch, partitionsConsumedConcurrently, stats, actionOnMessages);
+    if (producer) {
+        await producer.disconnect();
+    }
+    return ret;
 }
 
-async function runConsumeTransformProduce(brokers, consumeTopic, produceTopic, warmupMessages, totalMessageCnt, messageProcessTimeMs, ctpConcurrency) {
-    const kafka = new Kafka({
-        clientId: 'kafka-test-performance',
-        brokers: brokers.split(','),
-    });
+async function runConsumeTransformProduce(parameters, consumeTopic, produceTopic, warmupMessages, totalMessageCnt, messageProcessTimeMs, ctpConcurrency) {
+    const kafka = new Kafka(baseConfiguration(parameters));
 
     const producer = kafka.producer({});
     await producer.connect();
@@ -213,8 +256,9 @@ async function runConsumeTransformProduce(brokers, consumeTopic, produceTopic, w
                 if (messagesMeasured === totalMessageCnt) {
                     let elapsed = hrtime(startTime);
                     let durationNanos = elapsed[0] * 1e9 + elapsed[1];
+                    let durationSeconds = durationNanos / 1e9;
                     rate = (totalMessageSize / durationNanos) * 1e9 / (1024 * 1024); /* MB/s */
-                    console.log(`Recvd, transformed and sent ${messagesMeasured} messages, ${totalMessageSize} bytes; rate is ${rate} MB/s`);
+                    console.log(`Recvd, transformed and sent ${messagesMeasured} messages in ${durationSeconds}, ${totalMessageSize} bytes; rate is ${rate} MB/s`);
                     consumer.pause([{ topic }]);
                 }
             } else {
@@ -242,11 +286,8 @@ async function runConsumeTransformProduce(brokers, consumeTopic, produceTopic, w
 }
 
 
-async function runProducerConsumerTogether(brokers, topic, totalMessageCnt, msgSize, produceMessageProcessTimeMs, consumeMessageProcessTimeMs) {
-    const kafka = new Kafka({
-        clientId: 'kafka-test-performance',
-        brokers: brokers.split(','),
-    });
+async function runProducerConsumerTogether(parameters, topic, totalMessageCnt, msgSize, produceMessageProcessTimeMs, consumeMessageProcessTimeMs) {
+    const kafka = new Kafka(baseConfiguration(parameters));
 
     const producer = kafka.producer({});
     await producer.connect();
diff --git a/examples/performance/performance-primitives.js b/examples/performance/performance-primitives.js
index 8dce6b54..e1d66dd5 100644
--- a/examples/performance/performance-primitives.js
+++ b/examples/performance/performance-primitives.js
@@ -1,6 +1,14 @@
 const { Kafka, ErrorCodes, CompressionTypes } = require('../../').KafkaJS;
 const { randomBytes } = require('crypto');
 const { hrtime } = require('process');
+const {
+    runConsumer: runConsumerCommon,
+    runProducer: runProducerCommon,
+    runLagMonitoring: runLagMonitoringCommon,
+    genericProduceToTopic,
+    getAutoCommit,
+    PerformanceLogger,
+ } = require('./performance-primitives-common');
 
 module.exports = {
     runProducer,
@@ -8,23 +16,66 @@ module.exports = {
     runConsumeTransformProduce,
     runCreateTopics,
     runProducerConsumerTogether,
+    runLagMonitoring,
+    newCompatibleProducer,
 };
 
-async function runCreateTopics(brokers, topic, topic2) {
-    const kafka = new Kafka({
+
+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';
+const DEBUG = process.env.DEBUG;
+const STATISTICS_INTERVAL_MS = process.env.STATISTICS_INTERVAL_MS ? +process.env.STATISTICS_INTERVAL_MS : null;
+const ENABLE_LOGGING = DEBUG !== undefined || STATISTICS_INTERVAL_MS !== null;
+
+function baseConfiguration(parameters) {
+    let ret = {
         'client.id': 'kafka-test-performance',
-        "metadata.broker.list": brokers,
-    });
+        'metadata.broker.list': parameters.brokers,
+    };
+    if (parameters.securityProtocol &&
+        parameters.saslUsername &&
+        parameters.saslPassword) {
+        ret = {
+            ...ret,
+            'security.protocol': parameters.securityProtocol,
+            'sasl.mechanism': 'PLAIN',
+            'sasl.username': parameters.saslUsername,
+            'sasl.password': parameters.saslPassword,
+        };
+    }
+    if (DEBUG && !parameters.disableLogging) {
+        ret['debug'] = DEBUG;
+    }
+    if (parameters.logToFile) {
+        ret.kafkaJS = {
+            'logger': new PerformanceLogger(parameters.logToFile),
+        };
+    }
+    if (STATISTICS_INTERVAL_MS !== null) {
+        ret['statistics.interval.ms'] = STATISTICS_INTERVAL_MS;
+        ret['stats_cb'] = function (event) {
+            this.logger().info(event.message);
+        };
+    }
+    return ret;
+}
+
+async function runCreateTopics(parameters, topic, topic2, numPartitions) {
+    const adminParameters = {
+        ...parameters,
+        disableLogging: true,
+    };
+    const kafka = new Kafka(baseConfiguration(adminParameters));
 
     const admin = kafka.admin();
     await admin.connect();
 
     for (let t of [topic, topic2]) {
         let topicCreated = await admin.createTopics({
-            topics: [{ topic: t, numPartitions: 3 }],
+            topics: [{ topic: t, numPartitions }],
         }).catch(console.error);
         if (topicCreated) {
-            console.log(`Created topic ${t}`);
+            console.log(`Created topic ${t} with ${numPartitions} partitions`);
             continue;
         }
 
@@ -33,148 +84,166 @@ async function runCreateTopics(brokers, topic, topic2) {
         await new Promise(resolve => setTimeout(resolve, 1000)); /* Propagate. */
         await admin.createTopics({
             topics: [
-                { topic: t, numPartitions: 3 },
+                { topic: t, numPartitions },
             ],
         }).catch(console.error);
-        console.log(`Created topic ${t}`);
+        console.log(`Created topic ${t} with ${numPartitions} partitions`);
         await new Promise(resolve => setTimeout(resolve, 1000)); /* Propagate. */
     }
 
     await admin.disconnect();
 }
 
-async function runProducer(brokers, topic, batchSize, warmupMessages, totalMessageCnt, msgSize, compression) {
-    let totalMessagesSent = 0;
-    let totalBytesSent = 0;
+function runLagMonitoring(parameters, topic) {
+    const monitoringParameters = {
+        ...parameters,
+        disableLogging: true,
+    };
+    const kafka = new Kafka(baseConfiguration(monitoringParameters));
+    const admin = kafka.admin();
 
-    const message = {
-        value: randomBytes(msgSize),
-    }
+    return runLagMonitoringCommon(admin, topic);
+}
 
-    const messages = Array(batchSize).fill(message);
+class CompatibleProducer {
+    constructor(producer) {
+        this.producer = producer;
+    }
 
-    const kafka = new Kafka({
-        'client.id': 'kafka-test-performance',
-        'metadata.broker.list': brokers,
-        'compression.codec': CompressionTypes[compression],
-    });
+    async connect() {
+        return this.producer.connect();
+    }
 
-    const producer = kafka.producer();
-    await producer.connect();
+    async disconnect() {
+        return this.producer.disconnect();
+    }
 
-    console.log('Sending ' + warmupMessages + ' warmup messages.');
-    while (warmupMessages > 0) {
-        await producer.send({
-            topic,
-            messages,
-        });
-        warmupMessages -= batchSize;
+    isQueueFullError(err) {
+        return err.code === ErrorCodes.ERR__QUEUE_FULL;
     }
-    console.log('Sent warmup messages');
 
-    // Now that warmup is done, start measuring...
-    let startTime;
-    let promises = [];
-    startTime = hrtime();
-    let messagesDispatched = 0;
-
-    // The double while-loop allows us to send a bunch of messages and then
-    // await them all at once. We need the second while loop to keep sending
-    // in case of queue full errors, which surface only on awaiting.
-    while (totalMessageCnt == -1 || messagesDispatched < totalMessageCnt) {
-        while (totalMessageCnt == -1 || messagesDispatched < totalMessageCnt) {
-            promises.push(producer.send({
-                topic,
-                messages,
-            }).then(() => {
-                totalMessagesSent += batchSize;
-                totalBytesSent += batchSize * msgSize;
-            }).catch((err) => {
-                if (err.code === ErrorCodes.ERR__QUEUE_FULL) {
-                    /* do nothing, just send them again */
-                    messagesDispatched -= batchSize;
-                } else {
-                    console.error(err);
-                    throw err;
-                }
-            }));
-            messagesDispatched += batchSize;
-        }
-        await Promise.all(promises);
+    send(opts) {
+        return this.producer.send(opts);
     }
-    let elapsed = hrtime(startTime);
-    let durationNanos = elapsed[0] * 1e9 + elapsed[1];
-    let rate = (totalBytesSent / durationNanos) * 1e9 / (1024 * 1024); /* MB/s */
-    console.log(`Sent ${totalMessagesSent} messages, ${totalBytesSent} bytes; rate is ${rate} MB/s`);
+}
+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());
+}
 
-    await producer.disconnect();
-    return rate;
+async function runProducer(parameters, topic, batchSize, warmupMessages, totalMessageCnt, msgSize, compression, randomness, limitRPS) {
+    if (ENABLE_LOGGING && !parameters.logToFile) {
+        parameters.logToFile = './confluent-producer.log';
+    }
+    return runProducerCommon(newCompatibleProducer(parameters, compression), topic, batchSize, warmupMessages, totalMessageCnt, msgSize, compression, randomness, limitRPS);
 }
 
-async function runConsumer(brokers, topic, totalMessageCnt) {
-    const kafka = new Kafka({
-        'client.id': 'kafka-test-performance',
-        'metadata.broker.list': brokers,
-    });
+class CompatibleConsumer {
+    constructor(consumer) {
+        this.consumer = consumer;
+    }
 
-    const consumer = kafka.consumer({
-        'group.id': 'test-group' + Math.random(),
-        'enable.auto.commit': false,
-        'auto.offset.reset': 'earliest',
-        'fetch.queue.backoff.ms': '100',
-    });
-    await consumer.connect();
-    await consumer.subscribe({ topic });
+    async connect() {
+        return this.consumer.connect();
+    }
 
-    let messagesReceived = 0;
-    let messagesMeasured = 0;
-    let totalMessageSize = 0;
-    let startTime;
-    let rate;
-    const skippedMessages = 100;
+    async disconnect() {
+        return this.consumer.disconnect();
+    }
 
-    console.log("Starting consumer.");
+    async subscribe(opts) {
+        return this.consumer.subscribe(opts);
+    }
 
-    consumer.run({
-        eachMessage: async ({ topic, partition, message }) => {
-            messagesReceived++;
+    pause(topics) {
+        return this.consumer.pause(topics);
+    }
 
-            if (messagesReceived >= skippedMessages) {
-                messagesMeasured++;
-                totalMessageSize += message.value.length;
+    commitOffsetsOnBatchEnd(offsets) {
+        return this.consumer.commitOffsets(offsets);
+    }
 
-                if (messagesReceived === skippedMessages) {
-                    startTime = hrtime();
-                } else if (messagesMeasured === totalMessageCnt) {
-                    let elapsed = hrtime(startTime);
-                    let durationNanos = elapsed[0] * 1e9 + elapsed[1];
-                    rate = (totalMessageSize / durationNanos) * 1e9 / (1024 * 1024); /* MB/s */
-                    console.log(`Recvd ${messagesMeasured} messages, ${totalMessageSize} bytes; rate is ${rate} MB/s`);
-                    consumer.pause([{ topic }]);
-                }
-            }
-        }
-    });
+    run(opts) {
+        return this.consumer.run({
+            ...opts
+        });
+    }
+}
 
-    totalMessageSize = 0;
-    await new Promise((resolve) => {
-        let interval = setInterval(() => {
-            if (messagesMeasured >= totalMessageCnt) {
-                clearInterval(interval);
-                resolve();
-            }
-        }, 1000);
+function newCompatibleConsumer(parameters, eachBatch, messageSize, limitRPS) {
+    const minFetchBytes = messageSize * limitRPS;
+    const kafka = new Kafka(baseConfiguration(parameters));
+    const autoCommit = getAutoCommit();
+    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) {
+        groupId = 'test-group' + Math.random();
+    }
+    console.log(`New Confluent group id: ${groupId}`);
+    const consumer = kafka.consumer({
+        'group.id': groupId,
+        'auto.offset.reset': 'earliest',
+        'fetch.queue.backoff.ms': '100',
+        'fetch.min.bytes': minFetchBytes.toString(),
+        ...autoCommitOpts,
+        ...jsOpts,
+        ...higherLatencyClusterOpts,
     });
+    return new CompatibleConsumer(consumer);
+}
 
-    await consumer.disconnect();
-    return rate;
+
+async function runConsumer(parameters, topic, warmupMessages, totalMessageCnt, eachBatch, partitionsConsumedConcurrently, stats, produceToTopic, produceCompression, useCKJSProducerEverywhere,
+                           messageSize, limitRPS) {
+    let actionOnMessages = null;
+    let producer;
+    if (produceToTopic) {
+        const producerParameters = {
+            ...parameters,
+        };
+        if (ENABLE_LOGGING)
+            producerParameters.logToFile = './confluent-consumer-producer.log';
+        producer = newCompatibleProducer(producerParameters, produceCompression);
+        await producer.connect();
+        actionOnMessages = (messages) =>
+            genericProduceToTopic(producer, produceToTopic, messages);
+    }
+    const consumerParameters = {
+        ...parameters,
+    };
+    if (ENABLE_LOGGING)
+        consumerParameters.logToFile = eachBatch ? './confluent-consumer-batch.log' :
+                                                   './confluent-consumer-message.log';
+    const ret = await runConsumerCommon(
+        newCompatibleConsumer(consumerParameters, eachBatch, messageSize, limitRPS),
+        topic, warmupMessages, totalMessageCnt, eachBatch, partitionsConsumedConcurrently, stats, actionOnMessages);
+    if (producer) {
+        await producer.disconnect();
+    }
+    return ret;
 }
 
-async function runConsumeTransformProduce(brokers, consumeTopic, produceTopic, warmupMessages, totalMessageCnt, messageProcessTimeMs, ctpConcurrency) {
-    const kafka = new Kafka({
-        'client.id': 'kafka-test-performance',
-        'metadata.broker.list': brokers,
-    });
+async function runConsumeTransformProduce(parameters, consumeTopic, produceTopic, warmupMessages, totalMessageCnt, messageProcessTimeMs, ctpConcurrency) {
+    const kafka = new Kafka(baseConfiguration(parameters));
 
     const producer = kafka.producer({
         /* We want things to be flushed immediately as we'll be awaiting this. */
@@ -260,11 +329,8 @@ async function runConsumeTransformProduce(brokers, consumeTopic, produceTopic, w
     return rate;
 }
 
-async function runProducerConsumerTogether(brokers, topic, totalMessageCnt, msgSize, produceMessageProcessTimeMs, consumeMessageProcessTimeMs) {
-    const kafka = new Kafka({
-        'client.id': 'kafka-test-performance',
-        'metadata.broker.list': brokers,
-    });
+async function runProducerConsumerTogether(parameters, topic, totalMessageCnt, msgSize, produceMessageProcessTimeMs, consumeMessageProcessTimeMs) {
+    const kafka = new Kafka(baseConfiguration(parameters));
 
     const producer = kafka.producer({
         /* We want things to be flushed immediately as we'll be awaiting this. */
diff --git a/lib/kafkajs/_admin.js b/lib/kafkajs/_admin.js
index ed713546..a4ce1bac 100644
--- a/lib/kafkajs/_admin.js
+++ b/lib/kafkajs/_admin.js
@@ -174,6 +174,12 @@ class Admin {
     /* Delete properties which are already processed, or cannot be passed to node-rdkafka */
     delete rdKafkaConfig.kafkaJS;
 
+    if (Object.hasOwn(rdKafkaConfig, 'stats_cb')) {
+      if (typeof rdKafkaConfig['stats_cb'] === 'function')
+        this.#statsCb = rdKafkaConfig['stats_cb'];
+      delete rdKafkaConfig['stats_cb'];
+    }
+
     return rdKafkaConfig;
   }
 
@@ -203,6 +209,12 @@ class Admin {
     }
   }
 
+  /**
+   * Callback for the event.stats event, if defined.
+   * @private
+   */
+  #statsCb = null;
+
   /**
    * Set up the client and connect to the bootstrap brokers.
    * @returns {Promise<void>} Resolves when connection is complete, rejects on error.
@@ -225,6 +237,9 @@ class Admin {
             'ready': this.#readyCb.bind(this),
             'event.log': (msg) => loggerTrampoline(msg, this.#logger),
           });
+          if (this.#statsCb) {
+               this.#internalClient.on('event.stats',  this.#statsCb.bind(this));
+          }
         } else {
           const underlyingClient = this.#existingClient._getInternalClient();
           if (!underlyingClient) {
diff --git a/lib/kafkajs/_consumer.js b/lib/kafkajs/_consumer.js
index ada184f8..37b045de 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,37 @@ class Consumer {
     this.#cacheExpirationTimeoutMs = this.#maxPollIntervalMs;
     rdKafkaConfig['max.poll.interval.ms'] = this.#maxPollIntervalMs * 2;
 
+    if (Object.hasOwn(rdKafkaConfig, 'js.consumer.max.batch.size')) {
+      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 (Object.hasOwn(rdKafkaConfig, 'js.consumer.max.cache.size.per.worker.ms')) {
+      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'];
+    }
+
+    if (Object.hasOwn(rdKafkaConfig, 'stats_cb')) {
+      if (typeof rdKafkaConfig['stats_cb'] === 'function')
+        this.#statsCb = rdKafkaConfig['stats_cb'];
+      delete rdKafkaConfig['stats_cb'];
+    }
+
     return rdKafkaConfig;
   }
 
@@ -730,6 +776,13 @@ class Consumer {
     }
   }
 
+
+  /**
+   * Callback for the event.stats event, if defined.
+   * @private
+   */
+  #statsCb = null;
+
   /**
    * Converts headers returned by node-rdkafka into a format that can be used by the eachMessage/eachBatch callback.
    * @param {import("../..").MessageHeader[] | undefined} messageHeaders
@@ -761,7 +814,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 +838,7 @@ class Consumer {
       },
       heartbeat: async () => { /* no op */ },
       pause: this.pause.bind(this, [{ topic: message.topic, partitions: [message.partition] }]),
+      _worker: worker,
     };
   }
 
@@ -844,40 +898,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 +969,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 +985,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 +1048,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 +1066,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 +1136,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
@@ -1136,6 +1180,9 @@ class Consumer {
     this.#internalClient.on('error', this.#errorCb.bind(this));
     this.#internalClient.on('event.error', this.#errorCb.bind(this));
     this.#internalClient.on('event.log', (msg) => loggerTrampoline(msg, this.#logger));
+    if (this.#statsCb) {
+      this.#internalClient.on('event.stats', this.#statsCb.bind(this));
+    }
 
     return new Promise((resolve, reject) => {
       this.#connectPromiseFunc = { resolve, reject };
@@ -1262,12 +1309,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 +1370,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 +1439,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 +1493,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 +1501,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 +1590,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 +1611,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 +1630,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 +1659,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..d3130d1f 100644
--- a/lib/kafkajs/_producer.js
+++ b/lib/kafkajs/_producer.js
@@ -269,6 +269,12 @@ class Producer {
     /* TODO: Add a warning if dr_cb is set? Or else, create a trampoline for it. */
     rdKafkaConfig.dr_cb = true;
 
+    if (Object.hasOwn(rdKafkaConfig, 'stats_cb')) {
+      if (typeof rdKafkaConfig['stats_cb'] === 'function')
+        this.#statsCb = rdKafkaConfig['stats_cb'];
+      delete rdKafkaConfig['stats_cb'];
+    }
+
     return rdKafkaConfig;
   }
 
@@ -373,6 +379,12 @@ class Producer {
     this.#logger.error(err, this.#createProducerBindingMessageMetadata());
   }
 
+  /**
+   * Callback for the event.stats event, if defined.
+   * @private
+   */
+  #statsCb = null;
+
   /**
    * Set up the client and connect to the bootstrap brokers.
    *
@@ -394,6 +406,9 @@ class Producer {
     this.#internalClient.on('event.error', this.#errorCb.bind(this));
     this.#internalClient.on('error', this.#errorCb.bind(this));
     this.#internalClient.on('event.log', (msg) => loggerTrampoline(msg, this.#logger));
+    if (this.#statsCb) {
+      this.#internalClient.on('event.stats', this.#statsCb.bind(this));
+    }
 
     return new Promise((resolve, reject) => {
       this.#connectPromiseFunc = { resolve, reject };
@@ -622,27 +637,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..a8a9322f 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.7.0';
diff --git a/package-lock.json b/package-lock.json
index 45f6f0a3..e9a84eb5 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -1,12 +1,12 @@
 {
   "name": "@confluentinc/kafka-javascript",
-  "version": "1.6.0",
+  "version": "1.7.0",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "@confluentinc/kafka-javascript",
-      "version": "1.6.0",
+      "version": "1.7.0",
       "hasInstallScript": true,
       "license": "MIT",
       "workspaces": [
@@ -10077,7 +10077,7 @@
     },
     "schemaregistry": {
       "name": "@confluentinc/schemaregistry",
-      "version": "1.6.0",
+      "version": "1.7.0",
       "license": "MIT",
       "dependencies": {
         "@aws-sdk/client-kms": "^3.637.0",
diff --git a/package.json b/package.json
index 4d36d2da..1852a308 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@confluentinc/kafka-javascript",
-  "version": "1.6.0",
+  "version": "1.7.0",
   "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..b00c58f9 100644
--- a/schemaregistry/package.json
+++ b/schemaregistry/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@confluentinc/schemaregistry",
-  "version": "1.6.0",
+  "version": "1.7.0",
   "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;