When the consumer read timeout, try to read after the value of `timeout` #769

yunnysunny · 2020-03-22T05:25:28Z

The pull request is aimed to resolve the issue of #709 .

…ut`.

stale · 2020-07-01T13:21:15Z

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs.

giulliano-bueno · 2020-07-02T20:36:56Z

This PR seems to be good enough to be merged.

iradul · 2020-07-20T01:38:25Z

I think default value for this timeout is 1000 so this can be a breaking change for some. It would be much better to have a separate timeout field just for handling this scenario.

yunnysunny · 2020-07-21T09:10:36Z

Add a field of m_retry_read_ms to set the delay time before next read operation.

iradul · 2020-07-23T03:34:38Z

lib/kafka-consumer.js

@@ -17,7 +17,8 @@ var KafkaConsumerStream = require('./kafka-consumer-stream');
 var LibrdKafkaError = require('./error');
 var TopicPartition = require('./topic-partition');
 var shallowCopy = require('./util').shallowCopy;
-
+var DEFAULT_RETRY_READ_INTERVAL = 500;


Let's call this DEFAULT_CONSUME_LOOP_TIMEOUT_DELAY

iradul · 2020-07-23T03:34:50Z

lib/kafka-consumer.js

+ * Set the default value of waiting for the next time to read when current reading get none data.
+ * @param {number} intervalMs - number of milliseconds to sleep before the next reading
+ */
+KafkaConsumer.prototype.setDefaultRetryReadInterval = function(intervalMs) {


Please rename method name to setDefaultConsumeLoopTimeoutDelay

iradul · 2020-07-23T03:34:52Z

lib/kafka-consumer.js

+/**
+ * Set the default value of waiting for the next time to read when current reading get none data.
+ * @param {number} intervalMs - number of milliseconds to sleep before the next reading
+ */


Maybe reword like this

/** * Set the default sleep delay for the next consume loop after the previous one has timed out. * @param {number} intervalMs - number of milliseconds to sleep after a message fetch has timed out */

iradul · 2020-07-23T03:35:12Z

lib/kafka-consumer.js

+ * @param {number} intervalMs - number of milliseconds to sleep before the next reading
+ */
+KafkaConsumer.prototype.setDefaultRetryReadInterval = function(intervalMs) {
+  this._retryReadInterval = Number(intervalMs) || DEFAULT_RETRY_READ_INTERVAL;


Please rename _retryReadInterval to _consumeLoopTimeoutDelay

src/kafka-consumer.cc

iradul · 2020-07-23T03:36:46Z

src/workers.cc

@@ -420,10 +420,12 @@ void KafkaConsumerDisconnect::HandleErrorCallback() {

 KafkaConsumerConsumeLoop::KafkaConsumerConsumeLoop(Nan::Callback *callback,
                                     KafkaConsumer* consumer,
-                                     const int & timeout_ms) :
+                                     const int & timeout_ms,
+                                     const int & retry_interval_ms) :


Please rename to timeout_sleep_delay_ms

iradul · 2020-07-23T03:36:54Z

src/workers.h

@@ -288,6 +288,7 @@ class KafkaConsumerConsumeLoop : public MessageWorker {
  NodeKafka::KafkaConsumer * consumer;
  const int m_timeout_ms;
  unsigned int m_rand_seed;
+  const int m_retry_read_ms;


Please rename to m_timeout_sleep_delay_ms

iradul · 2020-07-23T03:39:29Z

lib/kafka-consumer.js

+ * @param {number} intervalMs - number of milliseconds to sleep before the next reading
+ */
+KafkaConsumer.prototype.setDefaultRetryReadInterval = function(intervalMs) {
+  this._retryReadInterval = Number(intervalMs) || DEFAULT_RETRY_READ_INTERVAL;


Also I think you should not wrap interval within Number(..) because validation is done in c++ code

iradul · 2020-07-23T03:45:44Z

lib/kafka-consumer.js

 }

 /**
 * Set the default consume timeout provided to c++land
 * @param {number} timeoutMs - number of milliseconds to wait for a message to be fetched
 */
 KafkaConsumer.prototype.setDefaultConsumeTimeout = function(timeoutMs) {
-  this._consumeTimeout = timeoutMs;
+  this._consumeTimeout = Number(timeoutMs) || DEFAULT_CONSUME_TIME_OUT;


Passing 0 here will not work, also you should not wrap interval within Number(..) because:
a) validation is done in c++ code
b) it's kind of breaking change because calling setDefaultConsumeTimeout('anything but number') will suddenly start working

Maybe something like this._consumeTimeout = timeoutMs === 0 ? timeoutMs : timeoutMs || DEFAULT_CONSUME_TIME_OUT; would be safer.

Set the value of this._consumeTimeout to timeoutMs directly.

iradul · 2020-07-25T16:01:16Z

lib/kafka-consumer.js

@@ -124,7 +125,8 @@ function KafkaConsumer(conf, topicConf) {
  this.globalConfig = conf;
  this.topicConfig = topicConf;

-  this._consumeTimeout = 1000;
+  this._consumeTimeout = DEFAULT_CONSUME_LOOP_TIMEOUT_DELAY;


Here you should use DEFAULT_CONSUME_TIME_OUT instead of DEFAULT_CONSUME_LOOP_TIMEOUT_DELAY

iradul · 2020-07-25T16:01:27Z

lib/kafka-consumer.js

@@ -369,7 +379,7 @@ KafkaConsumer.prototype.unsubscribe = function() {
 * is fetched.
 */
 KafkaConsumer.prototype.consume = function(number, cb) {
-  var timeoutMs = this._consumeTimeout || 1000;
+  var timeoutMs = this._consumeTimeout || DEFAULT_CONSUME_LOOP_TIMEOUT_DELAY;


Here you should use DEFAULT_CONSUME_TIME_OUT instead of DEFAULT_CONSUME_LOOP_TIMEOUT_DELAY

iradul · 2020-07-25T16:01:55Z

src/kafka-consumer.cc

-  if (!info[0]->IsNumber()) {
-    return Nan::ThrowError("Need to specify a timeout");
+  if (!info[1]->IsNumber()) {
+    return Nan::ThrowError("Need to specify a sleep delay");
  }


We need both info[0] and info[1]:

if (!info[0]->IsNumber()) { return Nan::ThrowError("Need to specify a timeout"); } if (!info[1]->IsNumber()) { return Nan::ThrowError("Need to specify a sleep delay"); }

iradul · 2020-07-26T15:06:39Z

Thanks!

yunnysunny · 2020-07-27T05:06:59Z

I'm sorry for taking some mistakes. I wrote the code during my overtime work when my head was not clear.

iradul · 2020-07-27T18:09:26Z

No worries, thanks again for PR!

When the consumer read timeout, try to read after the value of `timeo…

a50963c

…ut`.

yunnysunny closed this Mar 31, 2020

yunnysunny reopened this Mar 31, 2020

stale bot added the stale Stale issues label Jul 1, 2020

stale bot removed the stale Stale issues label Jul 2, 2020

Add filed m_retry_read_ms to the c++ class KafkaConsumerConsumeLoop

775231e

iradul reviewed Jul 23, 2020

View reviewed changes

src/kafka-consumer.cc Show resolved Hide resolved

iradul reviewed Jul 23, 2020

View reviewed changes

Change some variables' name about loop delay.

5f66cec

iradul reviewed Jul 25, 2020

View reviewed changes

Correct some assignment mistakes.

0b11c64

iradul merged commit c66b741 into Blizzard:master Jul 26, 2020

yunnysunny mentioned this pull request Oct 29, 2020

Consumer has 500ms delay #709

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

When the consumer read timeout, try to read after the value of `timeout` #769

When the consumer read timeout, try to read after the value of `timeout` #769

yunnysunny commented Mar 22, 2020

stale bot commented Jul 1, 2020

giulliano-bueno commented Jul 2, 2020

iradul commented Jul 20, 2020

yunnysunny commented Jul 21, 2020

iradul Jul 23, 2020 •

edited

Loading

iradul Jul 23, 2020

iradul Jul 23, 2020

iradul Jul 23, 2020

iradul Jul 23, 2020

iradul Jul 23, 2020

iradul Jul 23, 2020

iradul Jul 23, 2020

yunnysunny Jul 24, 2020

iradul Jul 25, 2020

iradul Jul 25, 2020

iradul Jul 25, 2020

iradul commented Jul 26, 2020

yunnysunny commented Jul 27, 2020

iradul commented Jul 27, 2020

When the consumer read timeout, try to read after the value of timeout #769

When the consumer read timeout, try to read after the value of timeout #769

Conversation

yunnysunny commented Mar 22, 2020

stale bot commented Jul 1, 2020

giulliano-bueno commented Jul 2, 2020

iradul commented Jul 20, 2020

yunnysunny commented Jul 21, 2020

iradul Jul 23, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

iradul commented Jul 26, 2020

yunnysunny commented Jul 27, 2020

iradul commented Jul 27, 2020

When the consumer read timeout, try to read after the value of `timeout` #769

When the consumer read timeout, try to read after the value of `timeout` #769

iradul Jul 23, 2020 •

edited

Loading