Fix bug where the timeout in the admin APIs wasn't taken into account. (#653)

Author: benjaminroux

lib/admin.js

@@ -119,11 +119,11 @@ AdminClient.prototype.createTopic = function(topic, timeout, cb) {
 
   if (typeof timeout === 'function') {
     cb = timeout;
-    timeout = 1000;
+    timeout = 5000;
   }
 
   if (!timeout) {
-    timeout = 1000;
+    timeout = 5000;
   }
 
   this._client.createTopic(topic, timeout, function(err) {
@@ -154,11 +154,11 @@ AdminClient.prototype.deleteTopic = function(topic, timeout, cb) {
 
   if (typeof timeout === 'function') {
     cb = timeout;
-    timeout = 1000;
+    timeout = 5000;
   }
 
   if (!timeout) {
-    timeout = 1000;
+    timeout = 5000;
   }
 
   this._client.deleteTopic(topic, timeout, function(err) {
@@ -181,7 +181,7 @@ AdminClient.prototype.deleteTopic = function(topic, timeout, cb) {
  * @param {string} topic - The topic to add partitions to, by name.
  * @param {number} totalPartitions - The total number of partitions the topic should have
  *                                   after the request
- * @param {number} timeout - Number of milliseconds to wait while trying to delete the topic.
+ * @param {number} timeout - Number of milliseconds to wait while trying to create the partitions.
  * @param {function} cb - The callback to be executed when finished
  */
 AdminClient.prototype.createPartitions = function(topic, totalPartitions, timeout, cb) {
@@ -191,11 +191,11 @@ AdminClient.prototype.createPartitions = function(topic, totalPartitions, timeou
 
   if (typeof timeout === 'function') {
     cb = timeout;
-    timeout = 1000;
+    timeout = 5000;
   }
 
   if (!timeout) {
-    timeout = 1000;
+    timeout = 5000;
   }
 
   this._client.createPartitions(topic, totalPartitions, timeout, function(err) {

src/admin.cc

@@ -9,6 +9,7 @@
 
 #include <string>
 #include <vector>
+#include <math.h>
 
 #include "src/workers.h"
 #include "src/admin.h"
@@ -151,23 +152,30 @@ v8::Local<v8::Object> AdminClient::NewInstance(v8::Local<v8::Value> arg) {
 rd_kafka_event_t* PollForEvent(
   rd_kafka_queue_t * topic_rkqu,
   rd_kafka_event_type_t event_type,
-  int max_tries,
   int timeout_ms) {
-  // Establish what attempt we are on
-  int attempt = 0;
+  // Initiate exponential timeout
+  int attempts = 1;
+  int exp_timeout_ms = timeout_ms;
+  if (timeout_ms > 2000) {
+    // measure optimal number of attempts
+    attempts = log10(timeout_ms / 1000) / log10(2) + 1;
+    // measure initial exponential timeout based on attempts
+    exp_timeout_ms = timeout_ms / (pow(2, attempts) - 1);
+  }
 
   rd_kafka_event_t * event_response = nullptr;
 
   // Poll the event queue until we get it
   do {
     // free previously fetched event
     rd_kafka_event_destroy(event_response);
-    // Increment attempt counter
-    attempt = attempt + 1;
-    event_response = rd_kafka_queue_poll(topic_rkqu, timeout_ms);
+    // poll and update attempts and exponential timeout
+    event_response = rd_kafka_queue_poll(topic_rkqu, exp_timeout_ms);
+    attempts = attempts - 1;
+    exp_timeout_ms = 2 * exp_timeout_ms;
   } while (
     rd_kafka_event_type(event_response) != event_type &&
-    attempt < max_tries);
+    attempts > 0);
 
   // If this isn't the type of response we want, or if we do not have a response
   // type, bail out with a null
@@ -204,8 +212,7 @@ Baton AdminClient::CreateTopic(rd_kafka_NewTopic_t* topic, int timeout_ms) {
     rd_kafka_event_t * event_response = PollForEvent(
       topic_rkqu,
       RD_KAFKA_EVENT_CREATETOPICS_RESULT,
-      5,
-      1000);
+      timeout_ms);
 
     // Destroy the queue since we are done with it.
     rd_kafka_queue_destroy(topic_rkqu);
@@ -284,8 +291,7 @@ Baton AdminClient::DeleteTopic(rd_kafka_DeleteTopic_t* topic, int timeout_ms) {
     rd_kafka_event_t * event_response = PollForEvent(
       topic_rkqu,
       RD_KAFKA_EVENT_DELETETOPICS_RESULT,
-      5,
-      1000);
+      timeout_ms);
 
     // Destroy the queue since we are done with it.
     rd_kafka_queue_destroy(topic_rkqu);
@@ -360,8 +366,7 @@ Baton AdminClient::CreatePartitions(
     rd_kafka_event_t * event_response = PollForEvent(
       topic_rkqu,
       RD_KAFKA_EVENT_CREATEPARTITIONS_RESULT,
-      5,
-      1000);
+      timeout_ms);
 
     // Destroy the queue since we are done with it.
     rd_kafka_queue_destroy(topic_rkqu);
@@ -480,7 +485,7 @@ NAN_METHOD(AdminClient::NodeCreateTopic) {
   AdminClient* client = ObjectWrap::Unwrap<AdminClient>(info.This());
 
   // Get the timeout
-  int timeout = Nan::To<int32_t>(info[2]).FromJust();
+  int timeout = Nan::To<int32_t>(info[1]).FromJust();
 
   std::string errstr;
   // Get that topic we want to create
@@ -524,7 +529,7 @@ NAN_METHOD(AdminClient::NodeDeleteTopic) {
     Nan::To<v8::String>(info[0]).ToLocalChecked());
 
   // Get the timeout
-  int timeout = Nan::To<int32_t>(info[2]).FromJust();
+  int timeout = Nan::To<int32_t>(info[1]).FromJust();
 
   // Get that topic we want to create
   rd_kafka_DeleteTopic_t* topic = rd_kafka_DeleteTopic_new(
@@ -563,6 +568,9 @@ NAN_METHOD(AdminClient::NodeCreatePartitions) {
   Nan::Callback *callback = new Nan::Callback(cb);
   AdminClient* client = ObjectWrap::Unwrap<AdminClient>(info.This());
 
+  // Get the timeout
+  int timeout = Nan::To<int32_t>(info[2]).FromJust();
+
   // Get the total number of desired partitions
   int partition_total_count = Nan::To<int32_t>(info[1]).FromJust();
 
@@ -585,7 +593,7 @@ NAN_METHOD(AdminClient::NodeCreatePartitions) {
 
   // Queue up dat work
   Nan::AsyncQueueWorker(new Workers::AdminClientCreatePartitions(
-    callback, client, new_partitions, 1000));
+    callback, client, new_partitions, timeout));
 
   return info.GetReturnValue().Set(Nan::Null());
 }