Add Producer poll from librdkafka background thread

Author: milindl

MIGRATION.md

@@ -181,6 +181,22 @@ producerRun().then(consumerRun).catch(console.error);
       messages: [ /* ... */ ],
     });
     ```
+  - It's recommended to send a number of messages without awaiting them, and then calling `flush` to ensure all messages are sent, rather than awaiting each message. This is more efficient.
+    Example:
+    ```javascript
+    const kafka = new Kafka({ kafkaJS: { /* ... */ }});
+    const producer = kafka.producer();
+    await producer.connect();
+    for (/*...*/) producer.send({ /* ... */});
+    await producer.flush({timeout: 5000});
+    ```
+
+    However, in case it is desired to await every message, `linger.ms` should be set to 0, to ensure that the default batching behaviour does not cause a delay in awaiting messages.
+    Example:
+    ```javascript
+    const kafka = new Kafka({ kafkaJS: { /* ... */ }});
+    const producer = kafka.producer({ 'linger.ms': 0 });
+    ```
 
 * A transactional producer (with a `transactionId`) set, **cannot** send messages without initiating a transaction using `producer.transaction()`.
 

lib/kafkajs/_producer.js

@@ -13,7 +13,6 @@ const { kafkaJSToRdKafkaConfig,
 const { Consumer } = require('./_consumer');
 const error = require('./_error');
 const { Buffer } = require('buffer');
-const { hrtime } = require('process');
 
 const ProducerState = Object.freeze({
   INIT: 0,
@@ -33,8 +32,6 @@ const CompressionTypes = Object.freeze({
   ZSTD: 'zstd',
 });
 
-const producerPollIntervalMs = 500;
-
 class Producer {
   /**
    * The config supplied by the user.
@@ -78,14 +75,6 @@ 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 = hrtime();
-
   /**
    * @constructor
    * @param {import("../../types/kafkajs").ProducerConfig} kJSConfig
@@ -277,16 +266,7 @@ class Producer {
 
     this.#state = ProducerState.CONNECTED;
 
-    /* Start a loop to poll. the queues. */
-    const pollInterval = setInterval(() => {
-      if (this.#state >= ProducerState.DISCONNECTING) {
-        clearInterval(pollInterval);
-        return;
-      }
-      this.#internalClient.poll();
-      this.#lastPollTime = hrtime();
-    }, producerPollIntervalMs);
-
+    this.#internalClient.setPollInBackground(true);
     this.#internalClient.on('delivery-report', this.#deliveryCallback.bind(this));
 
     // Resolve the promise.
@@ -564,16 +544,6 @@ class Producer {
           reject(err);
         }
       }));
-
-      /* 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 = hrtime(this.#lastPollTime);
-      const elapsedInNanos = elapsed[0] * 1e9 + elapsed[1];
-      if (elapsedInNanos > producerPollIntervalMs * 1000) {
-        this.#lastPollTime = hrtime();
-        this.#internalClient.poll();
-      }
-
     }
 
     /* The delivery report will be handled by the delivery-report event handler, and we can simply wait for it here. */

lib/producer.js

@@ -230,6 +230,23 @@ Producer.prototype.setPollInterval = function(interval) {
   return this;
 };
 
+/**
+ * Set automatic polling for events on the librdkafka background thread.
+ *
+ * This provides several advantages over `setPollInterval`, as the polling
+ * does not happen on the event loop, but on the C thread spawned by librdkafka,
+ * and can be more efficient for high-throughput producers.
+ *
+ * @param {boolean} set Whether to poll in the background or not.
+ * @note If set = true, this will disable any polling interval set by `setPollInterval`.
+ */
+Producer.prototype.setPollInBackground = function(set) {
+  if (set) {
+    this.setPollInterval(0); // Clear poll interval from JS.
+  }
+  this._client.setPollInBackground(set);
+}
+
 /**
  * Flush the producer
  *

src/producer.cc

@@ -33,7 +33,8 @@ namespace NodeKafka {
 Producer::Producer(Conf* gconfig, Conf* tconfig):
   Connection(gconfig, tconfig),
   m_dr_cb(),
-  m_partitioner_cb() {
+  m_partitioner_cb(),
+  m_is_background_polling(false) {
     std::string errstr;
 
     if (m_tconfig)
@@ -72,6 +73,7 @@ void Producer::Init(v8::Local<v8::Object> exports) {
   Nan::SetPrototypeMethod(tpl, "getMetadata", NodeGetMetadata);
   Nan::SetPrototypeMethod(tpl, "queryWatermarkOffsets", NodeQueryWatermarkOffsets);  // NOLINT
   Nan::SetPrototypeMethod(tpl, "poll", NodePoll);
+  Nan::SetPrototypeMethod(tpl, "setPollInBackground", NodeSetPollInBackground);
   Nan::SetPrototypeMethod(tpl, "setSaslCredentials", NodeSetSaslCredentials);
   Nan::SetPrototypeMethod(tpl, "setOAuthBearerToken", NodeSetOAuthBearerToken);
   Nan::SetPrototypeMethod(tpl, "setOAuthBearerTokenFailure",
@@ -330,9 +332,39 @@ Baton Producer::Produce(void* message, size_t size, std::string topic,
 }
 
 void Producer::Poll() {
+  // We're not allowed to call poll when we have forwarded the main
+  // queue to the background queue, as that would indirectly poll
+  // the background queue. However, that's not allowed by librdkafka.
+  if (m_is_background_polling) {
+    return;
+  }
   m_client->poll(0);
 }
 
+Baton Producer::SetPollInBackground(bool set) {
+  scoped_shared_read_lock lock(m_connection_lock);
+  rd_kafka_t* rk = this->m_client->c_ptr();
+  if (!IsConnected()) {
+    return Baton(RdKafka::ERR__STATE, "Producer is disconnected");
+  }
+
+  if (set && !m_is_background_polling) {
+    m_is_background_polling = true;
+    rd_kafka_queue_t* main_q = rd_kafka_queue_get_main(rk);
+    rd_kafka_queue_t* background_q = rd_kafka_queue_get_background(rk);
+    rd_kafka_queue_forward(main_q, background_q);
+    rd_kafka_queue_destroy(main_q);
+    rd_kafka_queue_destroy(background_q);
+  } else if (!set && m_is_background_polling) {
+    m_is_background_polling = false;
+    rd_kafka_queue_t* main_q = rd_kafka_queue_get_main(rk);
+    rd_kafka_queue_forward(main_q, NULL);
+    rd_kafka_queue_destroy(main_q);
+  }
+
+  return Baton(RdKafka::ERR_NO_ERROR);
+}
+
 void Producer::ConfigureCallback(const std::string &string_key, const v8::Local<v8::Function> &cb, bool add) {
   if (string_key.compare("delivery_cb") == 0) {
     if (add) {
@@ -686,6 +718,23 @@ NAN_METHOD(Producer::NodePoll) {
   }
 }
 
+NAN_METHOD(Producer::NodeSetPollInBackground) {
+  Nan::HandleScope scope;
+  if (info.Length() < 1 || !info[0]->IsBoolean()) {
+    // Just throw an exception
+    return Nan::ThrowError(
+        "Need to specify a boolean for setting or unsetting");
+  }
+  bool set = Nan::To<bool>(info[0]).FromJust();
+
+  Producer* producer = ObjectWrap::Unwrap<Producer>(info.This());
+  Baton b = producer->SetPollInBackground(set);
+  if (b.err() != RdKafka::ERR_NO_ERROR) {
+    return Nan::ThrowError(b.errstr().c_str());
+  }
+  info.GetReturnValue().Set(b.ToObject());
+}
+
 Baton Producer::Flush(int timeout_ms) {
   RdKafka::ErrorCode response_code;
   if (IsConnected()) {

src/producer.h

@@ -54,6 +54,7 @@ class Producer : public Connection {
   Baton Connect();
   void Disconnect();
   void Poll();
+  Baton SetPollInBackground(bool);
   #if RD_KAFKA_VERSION > 0x00090200
   Baton Flush(int timeout_ms);
   #endif
@@ -103,6 +104,7 @@ class Producer : public Connection {
   static NAN_METHOD(NodeConnect);
   static NAN_METHOD(NodeDisconnect);
   static NAN_METHOD(NodePoll);
+  static NAN_METHOD(NodeSetPollInBackground);
   #if RD_KAFKA_VERSION > 0x00090200
   static NAN_METHOD(NodeFlush);
   #endif
@@ -114,6 +116,7 @@ class Producer : public Connection {
 
   Callbacks::Delivery m_dr_cb;
   Callbacks::Partitioner m_partitioner_cb;
+  bool m_is_background_polling;
 };
 
 }  // namespace NodeKafka

types/rdkafka.d.ts

@@ -268,6 +268,7 @@ export class Producer extends Client<KafkaProducerEvents> {
     produce(topic: string, partition: NumberNullUndefined, message: MessageValue, key?: MessageKey, timestamp?: NumberNullUndefined, opaque?: any, headers?: MessageHeader[]): any;
 
     setPollInterval(interval: number): this;
+    setPollInBackground(set: boolean): void;
 
     static createWriteStream(conf: ProducerGlobalConfig, topicConf: ProducerTopicConfig, streamOptions: WriteStreamOptions): ProducerStream;