Don't modify passed messages in sendOptions

Author: emasab

ci/tests/run_perf_test.js

@@ -96,7 +96,7 @@ async function main() {
 
       await runCommand(`MODE=${mode} node performance-consolidated.js --create-topics`);
 
-      console.log(`Waiting 10s ms after topic creation before starting producer and consumers...`);
+      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...`);

lib/kafkajs/_producer.js

@@ -622,27 +622,32 @@ class Producer {
     const msgPromises = [];
     for (let i = 0; i < sendOptions.messages.length; i++) {
       const msg = sendOptions.messages[i];
-
-      if (!Object.hasOwn(msg, "partition") || msg.partition === null) {
-        msg.partition = -1;
+      let key = msg.key;
+      let value = msg.value;
+      let partition = msg.partition;
+      let headers = msg.headers;
+      let timestamp = msg.timestamp;
+
+      if (partition === undefined || msg.partition === null) {
+        partition = -1;
       }
 
-      if (typeof msg.value === 'string') {
-        msg.value = Buffer.from(msg.value);
+      if (typeof value === 'string') {
+        value = Buffer.from(value);
       }
 
-      if (Object.hasOwn(msg, "timestamp") && msg.timestamp) {
-        msg.timestamp = Number(msg.timestamp);
+      if (timestamp) {
+        timestamp = Number(timestamp);
       } else {
-        msg.timestamp = 0;
+        timestamp = 0;
       }
 
-      msg.headers = convertToRdKafkaHeaders(msg.headers);
+      headers = convertToRdKafkaHeaders(headers);
 
       msgPromises.push(new Promise((resolve, reject) => {
         const opaque = { resolve, reject };
         try {
-          this.#internalClient.produce(sendOptions.topic, msg.partition, msg.value, msg.key, msg.timestamp, opaque, msg.headers);
+          this.#internalClient.produce(sendOptions.topic, partition, value, key, timestamp, opaque, headers);
         } catch (err) {
           reject(err);
         }