Changed default value of maxWaitTimeInMs (#363)

* Changed default value of maxWaitTimeInMs

* Added test

* Add link

Author: PratRanj07

CHANGELOG.md

@@ -1,3 +1,11 @@
+# confluent-kafka-javascript 1.5.0
+
+v1.5.0 is a feature release. It is supported for all usage.
+
+## Fixes
+
+1. Fix issue of delay of up to 5s in receiving messages after pause and resume, or seek (#285, #363).
+
 # confluent-kafka-javascript v1.4.1
 
 v1.4.1 is a maintenance release. It is supported for all usage.

MIGRATION.md

@@ -222,6 +222,7 @@ producerRun().then(consumerRun).catch(console.error);
   | allowAutoTopicCreation   | true                              | Determines if a topic should be created if it doesn't exist while consuming.                                                                                                                                                                |
   | **maxBytesPerPartition** | 1048576 (1MB)                     | determines how many bytes can be fetched in one request from a single partition. There is a change in semantics, this size grows dynamically if a single message larger than this is encountered, and the client does not get stuck.        |
   | minBytes                 | 1                                 | Minimum number of bytes the broker responds with (or wait until `maxWaitTimeInMs`)                                                                                                                                                          |
+  | **maxWaitTimeInMs**      | 500                               | Maximum time the broker may wait to fill the Fetch response with minBytes of messages. Its default value has been changed to match librdkafka's configuration.
   | maxBytes                 | 10485760 (10MB)                   | Maximum number of bytes the broker responds with.                                                                                                                                                                                           |
   | **retry**                | object                            | Identical to `retry` in the common configuration. This takes precedence over the common config retry.                                                                                                                                       |
   | readUncommitted          | false                             | If true, consumer will read transactional messages which have not been committed.                                                                                                                                                           |

lib/kafkajs/_consumer.js

@@ -563,8 +563,6 @@ class Consumer {
 
     if (Object.hasOwn(kjsConfig, 'maxWaitTimeInMs')) {
       rdKafkaConfig['fetch.wait.max.ms'] = kjsConfig.maxWaitTimeInMs;
-    } else {
-      rdKafkaConfig['fetch.wait.max.ms'] = 5000;
     }
 
     if (Object.hasOwn(kjsConfig, 'minBytes')) {

test/promisified/consumer/pauseSeekResume.spec.js

@@ -0,0 +1,77 @@
+/* eslint-disable no-unused-vars */
+jest.setTimeout(30000);
+
+const process = require('process');
+const {
+    secureRandom,
+    createTopic,
+    waitFor,
+    createProducer,
+    createConsumer,
+    sleep,
+} = require('../testhelpers');
+
+// See https://github.com/confluentinc/confluent-kafka-javascript/issues/285
+describe('Consumer commit', () => {
+    let topicName, groupId, producer, consumer;
+
+    beforeEach(async () => {
+        topicName = `test-topic-${secureRandom()}`;
+        groupId = `consumer-group-id-${secureRandom()}`;
+        await createTopic({ topic: topicName, partitions: 1 });
+        producer = createProducer({});
+        consumer = createConsumer({
+            groupId,
+            fromBeginning: true,
+            autoCommit: false,
+        });
+    });
+
+    afterEach(async () => {
+        if (consumer) await consumer.disconnect();
+        if (producer) await producer.disconnect();
+    });
+
+    it('should not have significant latency after pause-seek-resume', async () => {
+        const initialMessages = Array(30).fill().map(() => ({ value: `value-${secureRandom()}` }));
+        await producer.connect();
+        await producer.send({ topic: topicName, messages: initialMessages });
+        await producer.flush();
+
+        let msgCount = 0;
+        let maxLatencyMs = 0;
+        let startTime = process.hrtime.bigint();
+
+        await consumer.connect();
+        await consumer.subscribe({ topic: topicName });
+
+        await consumer.run({
+            eachMessage: async ({ topic, partition, message }) => {
+                const endTime = process.hrtime.bigint();
+                const latencyMs = Number((endTime - startTime) / BigInt(1e6));
+                if (msgCount > 0) maxLatencyMs = Math.max(maxLatencyMs, latencyMs);
+
+                msgCount++;
+                await sleep(10);
+                startTime = process.hrtime.bigint();
+            }
+        });
+
+        // Wait for 5 messages to be processed
+        await waitFor(() => msgCount >= 5, () => null, { delay: 100, timeout: 10000 });
+
+        await consumer.pause([{ topic: topicName, partition: 0 }]);
+        await consumer.seek({ topic: topicName, partition: 0, offset: '0' });
+        await consumer.resume([{ topic: topicName, partition: 0 }]);
+        startTime = process.hrtime.bigint(); // Reset timer after resume
+
+        // Wait for all messages to be processed (30 original + 5 after seek)
+        await waitFor(() => msgCount >= 35, () => null, { delay: 100, timeout: 10000 });
+
+        // Assert latency is reasonable (e.g., < 4000ms)
+        expect(maxLatencyMs).toBeLessThan(4000);
+
+        // Assert all messages processed
+        expect(msgCount).toBeGreaterThanOrEqual(35);
+    });
+});