Node-rdkafka: Synchronous .produce

[corybill]

I was wondering if someone could explain to me the synchronous nature of the produce function and why it is synchronous. Let's take an example. If I were to set 'request.required.acks' to -1 (i.e. write to all replicas before responding), does that mean that the produce function will be blocking until kafka returns with a success?

By looking at the code I see one other possibility. That is you are synchronously writing to a write stream, and then relying on the 'delivery-report' event to notify of actual success. If this is the case, then I can just provide a function in the 'dr_cb' parameter when creating the producer, correct? This callback will let me know the result from the kafka response.

Using a callback that is global to a Producer instance instead of a callback for a single request means that I cannot link a success message from a 'delivery-report' back to the specific call chain that initiated the request. Am I correct? If I am wrong here, an example would be great on how this can be achieved.

Can someone please explain to me when the this driver will give control back to the user? i.e., What does the synchronous call do? Thanks again for your help. I really appreciate someone taking the time to clear this up.

EDIT: One more question. The new producer uses 'acks' in place of 'request.required.acks'. Does the C++ module use the old producer (v8 or earlier) or is the documentation just out of date?

[webmakersteve]

The reason produce is synchronous is because it is a non blocking call. What it does is add your message to librdkafka's internal queue buffer. Then, those messages are sent a time later when librdkafka produces its batch of messages.

The delivery-report event is intended to tell you that a delivery succeeded or failed. You can correlate the message that failed or succeeded using the information passed to your callback. You can correlate the message currently using the payload, but I am going to add an opaque that you can use instead in a future release.

producer.on('delivery-report', (err, report) => {
const payload = report.value;
});

For the payload to show up, you will need to add dr_msg_cb: true to your producer config.

acks is an alias to request.required.acks. You can use either one and it should work.

[webmakersteve]

98 Adds support for using opaques like so

producer.produce(topic', partition, message, key, { messageId: 20 });

producer.on('delivery-report', (err, report) => {
   if (report.opaque && report.opaque.messageId === 20) {
      // we got it
   }
});