Add Producer flush

Author: milindl

lib/kafkajs/_producer.js

@@ -679,6 +679,61 @@ class Producer {
 
     this.#internalClient.setSaslCredentials(args.username, args.password);
   }
+
+  /**
+   * Flushes any pending messages.
+   *
+   * Messages are batched internally by librdkafka for performance reasons.
+   * Continously sent messages are batched upto a timeout, or upto a maximum
+   * size. Calling flush sends any pending messages immediately without
+   * waiting for this size or timeout.
+   *
+   * @param {number} args.timeout Time to try flushing for in milliseconds.
+   * @returns {Promise<void>} Resolves on successful flush.
+   * @throws {KafkaJSTimeout} if the flush times out.
+   *
+   * @note This is only useful when using asynchronous sends.
+   *       For example, the following code does not get any benefit from flushing,
+   *       since `await`ing the send waits for the delivery report, and the message
+   *       has already been sent by the time we start flushing:
+   *         for (let i = 0; i < 100; i++) await send(...);
+   *         await flush(...) // Not useful.
+   *
+   *       However, using the following code may put these 5 messages into a batch
+   *       and then the subsequent `flush` will send the batch altogether (as long as
+   *       batch size, etc. are conducive to batching):
+   *         for (let i = 0; i < 5; i++) send(...);
+   *         await flush({timeout: 5000});
+   */
+  async flush(args = { timeout: 500 }) {
+    if (this.#state !== ProducerState.CONNECTED) {
+      throw new error.KafkaJSError("Cannot flush without awaiting connect()", { code: error.ErrorCodes.ERR__STATE });
+    }
+
+    if (!Object.hasOwn(args, 'timeout')) {
+      throw new error.KafkaJSError("timeout must be set for flushing", { code: error.ErrorCodes.ERR__INVALID_ARG });
+    }
+
+    return new Promise((resolve, reject) => {
+      this.#internalClient.flush(args.timeout, (err) => {
+        if (err) {
+          const kjsErr = createKafkaJsErrorFromLibRdKafkaError(err);
+          if (err.code === error.ErrorCodes.ERR__TIMED_OUT) {
+            /* See reason below for yield. Same here - but for partially processed delivery reports. */
+            setTimeout(() => reject(kjsErr), 0);
+          } else {
+            reject(kjsErr);
+          }
+          return;
+        }
+        /* Yielding here allows any 'then's and 'awaits' on associated sends to be scheduled
+         * before flush completes, which means that the user doesn't have to yield themselves.
+         * It's not necessary that all the 'then's and 'awaits' will be able to run, but
+         * it's better than nothing. */
+        setTimeout(resolve, 0);
+      });
+    });
+  }
 }
 
 module.exports = { Producer, CompressionTypes };

src/workers.h

@@ -36,8 +36,13 @@ class ErrorAwareWorker : public Nan::AsyncWorker {
   void HandleErrorCallback() {
     Nan::HandleScope scope;
 
+    // Construct error and add code to it.
+    v8::Local<v8::Value> error = Nan::Error(ErrorMessage());
+    Nan::Set(error.As<v8::Object>(), Nan::New("code").ToLocalChecked(),
+      Nan::New(GetErrorCode()));
+
     const unsigned int argc = 1;
-    v8::Local<v8::Value> argv[argc] = { Nan::Error(ErrorMessage()) };
+    v8::Local<v8::Value> argv[argc] = { error };
 
     callback->Call(argc, argv);
   }

test/promisified/producer/flush.spec.js

@@ -0,0 +1,88 @@
+jest.setTimeout(10000);
+
+const {
+    secureRandom,
+    createProducer,
+    createTopic,
+} = require('../testhelpers');
+const { Kafka } = require('../../../lib').KafkaJS;
+const process = require('process');
+
+describe('Producer > Flush', () => {
+    let producer, topicName, message;
+
+    beforeEach(async () => {
+        producer = createProducer({
+        }, {
+            'linger.ms': 5000, /* large linger ms to test flush */
+            'queue.buffering.max.kbytes': 2147483647, /* effectively unbounded */
+        });
+
+        topicName = `test-topic-${secureRandom()}`;
+        message = { key: `key-${secureRandom()}`, value: `value-${secureRandom()}` };
+
+        await createTopic({ topic: topicName });
+    })
+
+    afterEach(async () => {
+        producer && (await producer.disconnect());
+    })
+
+
+    it('does not wait for linger.ms',
+        async () => {
+            await producer.connect();
+            let messageSent = false;
+            const startTime = process.hrtime();
+            let diffTime;
+
+            producer.send({ topic: topicName, messages: [message] }).then(() => {
+                messageSent = true;
+                diffTime = process.hrtime(startTime);
+            });
+
+            await producer.flush({ timeout: 5000 });
+            expect(messageSent).toBe(true);
+
+            const diffTimeSeconds = diffTime[0] + diffTime[1] / 1e9;
+            expect(diffTimeSeconds).toBeLessThan(5);
+        }
+    );
+
+    it('does not matter when awaiting sends',
+        async () => {
+            await producer.connect();
+            let messageSent = false;
+            const startTime = process.hrtime();
+            let diffTime;
+
+            await producer.send({ topic: topicName, messages: [message] }).then(() => {
+                messageSent = true;
+                diffTime = process.hrtime(startTime);
+            });
+
+            await producer.flush({ timeout: 1000 });
+            expect(messageSent).toBe(true);
+
+            const diffTimeSeconds = diffTime[0] + diffTime[1] / 1e9;
+            expect(diffTimeSeconds).toBeGreaterThan(5);
+        }
+    );
+
+    it('times out if messages are pending',
+        async () => {
+            await producer.connect();
+            let messageSent = false;
+
+            /* Larger number of messages */
+            producer.send({ topic: topicName, messages: Array(100).fill(message) }).then(() => {
+                messageSent = true;
+            });
+
+            /* Small timeout */
+            await expect(producer.flush({ timeout: 1 })).rejects.toThrow(Kafka.KafkaJSTimeout);
+            expect(messageSent).toBe(false);
+        }
+    );
+
+})