Stop consume loop thread when disconnecting (#1017)

package.json

@@ -1,6 +1,6 @@
 {
   "name": "node-rdkafka",
-  "version": "v2.16.0",
+  "version": "v2.16.1",
   "description": "Node.js bindings for librdkafka",
   "librdkafka": "2.1.1",
   "main": "lib/index.js",

src/kafka-consumer.cc

@@ -92,11 +92,6 @@ Baton KafkaConsumer::Disconnect() {
     }
   }
 
-  if (m_consume_loop != nullptr) {
-    delete m_consume_loop;
-    m_consume_loop = nullptr;
-  }
-
   m_is_closing = false;
 
   return Baton(err);
@@ -1192,6 +1187,18 @@ NAN_METHOD(KafkaConsumer::NodeDisconnect) {
   Nan::Callback *callback = new Nan::Callback(cb);
   KafkaConsumer* consumer = ObjectWrap::Unwrap<KafkaConsumer>(info.This());
 
+  Workers::KafkaConsumerConsumeLoop* consumeLoop = (Workers::KafkaConsumerConsumeLoop*)consumer->m_consume_loop;
+  if (consumeLoop != nullptr) {
+    // stop the consume loop
+    consumeLoop->Close();
+
+    // cleanup the async worker
+    consumeLoop->WorkComplete();
+    consumeLoop->Destroy();
+
+    consumer->m_consume_loop = nullptr;
+  }
+
   Nan::AsyncQueueWorker(
     new Workers::KafkaConsumerDisconnect(callback, consumer));
   info.GetReturnValue().Set(Nan::Null());

src/kafka-consumer.h

@@ -98,7 +98,7 @@ class KafkaConsumer : public Connection {
   int m_partition_cnt;
   bool m_is_subscribed = false;
 
-  void* m_consume_loop;
+  void* m_consume_loop = nullptr;
 
   // Node methods
   static NAN_METHOD(NodeConnect);

src/workers.cc

@@ -663,13 +663,19 @@ KafkaConsumerConsumeLoop::KafkaConsumerConsumeLoop(Nan::Callback *callback,
                                      const int & timeout_sleep_delay_ms) :
   MessageWorker(callback),
   consumer(consumer),
+  m_looping(true),
   m_timeout_ms(timeout_ms),
   m_timeout_sleep_delay_ms(timeout_sleep_delay_ms) {
   uv_thread_create(&thread_event_loop, KafkaConsumerConsumeLoop::ConsumeLoop, (void*)this);
 }
 
 KafkaConsumerConsumeLoop::~KafkaConsumerConsumeLoop() {}
 
+void KafkaConsumerConsumeLoop::Close() {
+  m_looping = false;
+  uv_thread_join(&thread_event_loop);
+}
+
 void KafkaConsumerConsumeLoop::Execute(const ExecutionMessageBus& bus) {
   // ConsumeLoop is used instead
 }
@@ -680,8 +686,7 @@ void KafkaConsumerConsumeLoop::ConsumeLoop(void *arg) {
   KafkaConsumer* consumer = consumerLoop->consumer;
 
   // Do one check here before we move forward
-  bool looping = true;
-  while (consumer->IsConnected() && looping) {
+  while (consumerLoop->m_looping && consumer->IsConnected()) {
     Baton b = consumer->Consume(consumerLoop->m_timeout_ms);
     RdKafka::ErrorCode ec = b.err();
     if (ec == RdKafka::ERR_NO_ERROR) {
@@ -711,15 +716,15 @@ void KafkaConsumerConsumeLoop::ConsumeLoop(void *arg) {
         default:
           // Unknown error. We need to break out of this
           consumerLoop->SetErrorBaton(b);
-          looping = false;
+          consumerLoop->m_looping = false;
           break;
         }
     } else if (ec == RdKafka::ERR_UNKNOWN_TOPIC_OR_PART || ec == RdKafka::ERR_TOPIC_AUTHORIZATION_FAILED) {
       bus.SendWarning(ec);
     } else {
       // Unknown error. We need to break out of this
       consumerLoop->SetErrorBaton(b);
-      looping = false;
+      consumerLoop->m_looping = false;
     }
   }
 }

src/workers.h

@@ -373,6 +373,7 @@ class KafkaConsumerConsumeLoop : public MessageWorker {
   ~KafkaConsumerConsumeLoop();
 
   static void ConsumeLoop(void *arg);
+  void Close();
   void Execute(const ExecutionMessageBus&);
   void HandleOKCallback();
   void HandleErrorCallback();
@@ -383,6 +384,7 @@ class KafkaConsumerConsumeLoop : public MessageWorker {
   const int m_timeout_ms;
   unsigned int m_rand_seed;
   const int m_timeout_sleep_delay_ms;
+  bool m_looping;
 };
 
 class KafkaConsumerConsume : public ErrorAwareWorker {