Poll producer in send() loops

Author: milindl

CONTRIBUTING.md

@@ -192,6 +192,16 @@ gdb node
 
 You can add breakpoints and so on after that.
 
+### Debugging and Profiling JavaScript
+
+Run the code with the `--inspect` flag, and then open `chrome://inspect` in Chrome and connect to the debugger.
+
+Example:
+
+```
+node --inspect path/to/file.js
+```
+
 ## Updating librdkafka version
 
 The librdkafka should be periodically updated to the latest release in https://github.com/confluentinc/librdkafka/releases

lib/kafkajs/_producer.js

@@ -25,12 +25,14 @@ const ProducerState = Object.freeze({
 });
 
 const CompressionTypes = Object.freeze({
-  None: 'none',
+  NONE: 'none',
   GZIP: 'gzip',
   SNAPPY: 'snappy',
   LZ4: 'lz4',
   ZSTD: 'zstd',
-})
+});
+
+const producerPollIntervalMs = 500;
 
 class Producer {
   /**
@@ -75,6 +77,14 @@ class Producer {
    */
   #logger = new DefaultLogger();
 
+  /**
+   * Stores the time of the last poll.
+   * In case we are producing in a tight loop, the interval timer will not
+   * fire, and we won't poll. By maintaining the last poll time, we can
+   * poll at the end of send() and sendBatch().
+   */
+  #lastPollTime = process.hrtime();
+
   /**
    * @constructor
    * @param {import("../../types/kafkajs").ProducerConfig} kJSConfig
@@ -273,7 +283,8 @@ class Producer {
         return;
       }
       this.#internalClient.poll();
-    }, 500);
+      this.#lastPollTime = process.hrtime();
+    }, producerPollIntervalMs);
 
     this.#internalClient.on('delivery-report', this.#deliveryCallback.bind(this));
 
@@ -541,6 +552,15 @@ class Producer {
         }
       }));
 
+      /* Poll if we haven't polled in a while. This can be the case if we're producing
+      * in a tight loop without awaiting the produce. */
+      const elapsed = process.hrtime(this.#lastPollTime);
+      const elapsedInNanos = elapsed[0] * 1e9 + elapsed[1];
+      if (elapsedInNanos > producerPollIntervalMs * 1000) {
+        this.#lastPollTime = process.hrtime();
+        this.#internalClient.poll();
+      }
+
     }
 
     /* The delivery report will be handled by the delivery-report event handler, and we can simply wait for it here. */