Handle consumer topic errors in librdkafka v1.5.0+ (#864)

Author: iradul

README.md

@@ -504,6 +504,7 @@ The following table lists events for this API.
 |-------|----------|
 |`data` | When using the Standard API consumed messages are emitted in this event. |
 |`partition.eof` | When using Standard API and the configuration option `enable.partition.eof` is set, `partition.eof` events are emitted in this event. The event contains `topic`, `partition` and `offset` properties. |
+|`warning` | The event is emitted in case of `UNKNOWN_TOPIC_OR_PART` or `TOPIC_AUTHORIZATION_FAILED` errors when consuming in *Flowing mode*. Since the consumer will continue working if the error is still happening, the warning event should reappear after the next metadata refresh. To control the metadata refresh rate set `topic.metadata.refresh.interval.ms` property. Once you resolve the error, you can manually call `getMetadata` to speed up consumer recovery. |
 |`disconnected` | The `disconnected` event is emitted when the broker disconnects. <br><br>This event is only emitted when `.disconnect` is called. The wrapper will always try to reconnect otherwise. |
 |`ready` | The `ready` event is emitted when the `Consumer` is ready to read messages. |
 |`event` | The `event` event is emitted when `librdkafka` reports an event (if you opted in via the `event_cb` option).|

e2e/both.spec.js

@@ -228,7 +228,7 @@ describe('Consumer/Producer', function() {
 
       consumer.once('partition.eof', function(eof) {
         events.push("partition.eof");
-      })
+      });
 
       setTimeout(function() {
         producer.produce(topic, null, buffer, null);
@@ -362,8 +362,23 @@ describe('Consumer/Producer', function() {
             startOffset + 2 ]);
         done();
       }, 6000);
-    })
-  })
+    });
+  });
+
+  it('should emit [warning] event on UNKNOWN_TOPIC_OR_PART error: consumeLoop', function(done) {
+    consumer.on('warning', function (err) {
+      if (err.code === Kafka.CODES.ERRORS.ERR_UNKNOWN_TOPIC_OR_PART) {
+        consumer.disconnect(function() {
+          done();
+        });
+      } else {
+        t.ifError(err);
+      }
+    });
+
+    consumer.subscribe(['non_existing_topic']);
+    consumer.consume();
+  });
 
   it('should be able to produce and consume messages with one header value as string: consumeLoop', function(done) {
     var headers = [

lib/kafka-consumer.js

@@ -426,7 +426,7 @@ KafkaConsumer.prototype.consume = function(number, cb) {
 KafkaConsumer.prototype._consumeLoop = function(timeoutMs, cb) {
   var self = this;
   var retryReadInterval = this._consumeLoopTimeoutDelay;
-  self._client.consumeLoop(timeoutMs, retryReadInterval, function readCallback(err, message, eofEvent) {
+  self._client.consumeLoop(timeoutMs, retryReadInterval, function readCallback(err, message, eofEvent, warning) {
 
     if (err) {
       // A few different types of errors here
@@ -436,6 +436,8 @@ KafkaConsumer.prototype._consumeLoop = function(timeoutMs, cb) {
       cb(LibrdKafkaError.create(err));
     } else if (eofEvent) {
       self.emit('partition.eof', eofEvent);
+    } else if (warning) {
+      self.emit('warning', LibrdKafkaError.create(warning));
     } else {
       /**
        * Data event. called whenever a message is received.

src/workers.cc

@@ -435,7 +435,8 @@ void KafkaConsumerConsumeLoop::Execute(const ExecutionMessageBus& bus) {
   bool looping = true;
   while (consumer->IsConnected() && looping) {
     Baton b = consumer->Consume(m_timeout_ms);
-    if (b.err() == RdKafka::ERR_NO_ERROR) {
+    RdKafka::ErrorCode ec = b.err();
+    if (ec == RdKafka::ERR_NO_ERROR) {
       RdKafka::Message *message = b.data<RdKafka::Message*>();
       switch (message->err()) {
         case RdKafka::ERR__PARTITION_EOF:
@@ -472,6 +473,8 @@ void KafkaConsumerConsumeLoop::Execute(const ExecutionMessageBus& bus) {
           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
       SetErrorBaton(b);
@@ -480,36 +483,43 @@ void KafkaConsumerConsumeLoop::Execute(const ExecutionMessageBus& bus) {
   }
 }
 
-void KafkaConsumerConsumeLoop::HandleMessageCallback(RdKafka::Message* msg) {
+void KafkaConsumerConsumeLoop::HandleMessageCallback(RdKafka::Message* msg, RdKafka::ErrorCode ec) {
   Nan::HandleScope scope;
 
-  const unsigned int argc = 3;
+  const unsigned int argc = 4;
   v8::Local<v8::Value> argv[argc];
 
   argv[0] = Nan::Null();
-  switch (msg->err()) {
-    case RdKafka::ERR__PARTITION_EOF: {
-      argv[1] = Nan::Null();
-      v8::Local<v8::Object> eofEvent = Nan::New<v8::Object>();
-
-      Nan::Set(eofEvent, Nan::New<v8::String>("topic").ToLocalChecked(),
-        Nan::New<v8::String>(msg->topic_name()).ToLocalChecked());
-      Nan::Set(eofEvent, Nan::New<v8::String>("offset").ToLocalChecked(),
-        Nan::New<v8::Number>(msg->offset()));
-      Nan::Set(eofEvent, Nan::New<v8::String>("partition").ToLocalChecked(),
-        Nan::New<v8::Number>(msg->partition()));
-
-      argv[2] = eofEvent;
-      break;
+  if (msg == NULL) {
+    argv[1] = Nan::Null();
+    argv[2] = Nan::Null();
+    argv[3] = Nan::New<v8::Number>(ec);
+  } else {
+    argv[3] = Nan::Null();
+    switch (msg->err()) {
+      case RdKafka::ERR__PARTITION_EOF: {
+        argv[1] = Nan::Null();
+        v8::Local<v8::Object> eofEvent = Nan::New<v8::Object>();
+
+        Nan::Set(eofEvent, Nan::New<v8::String>("topic").ToLocalChecked(),
+          Nan::New<v8::String>(msg->topic_name()).ToLocalChecked());
+        Nan::Set(eofEvent, Nan::New<v8::String>("offset").ToLocalChecked(),
+          Nan::New<v8::Number>(msg->offset()));
+        Nan::Set(eofEvent, Nan::New<v8::String>("partition").ToLocalChecked(),
+          Nan::New<v8::Number>(msg->partition()));
+
+        argv[2] = eofEvent;
+        break;
+      }
+      default:
+        argv[1] = Conversion::Message::ToV8Object(msg);
+        argv[2] = Nan::Null();
+        break;
     }
-    default:
-      argv[1] = Conversion::Message::ToV8Object(msg);
-      argv[2] = Nan::Null();
-      break;
-  }
 
-  // We can delete msg now
-  delete msg;
+    // We can delete msg now
+    delete msg;
+  }
 
   callback->Call(argc, argv);
 }

src/workers.h

@@ -90,20 +90,26 @@ class MessageWorker : public ErrorAwareWorker {
     }
 
     std::vector<RdKafka::Message*> message_queue;
+    std::vector<RdKafka::ErrorCode> warning_queue;
 
     {
       scoped_mutex_lock lock(m_async_lock);
       // Copy the vector and empty it
       m_asyncdata.swap(message_queue);
+      m_asyncwarning.swap(warning_queue);
     }
 
     for (unsigned int i = 0; i < message_queue.size(); i++) {
-      HandleMessageCallback(message_queue[i]);
+      HandleMessageCallback(message_queue[i], RdKafka::ERR_NO_ERROR);
 
       // we are done with it. it is about to go out of scope
       // for the last time so let's just free it up here. can't rely
       // on the destructor
     }
+
+    for (unsigned int i = 0; i < warning_queue.size(); i++) {
+      HandleMessageCallback(NULL, warning_queue[i]);
+    }
   }
 
   class ExecutionMessageBus {
@@ -112,13 +118,16 @@ class MessageWorker : public ErrorAwareWorker {
      void Send(RdKafka::Message* m) const {
        that_->Produce_(m);
      }
+     void SendWarning(RdKafka::ErrorCode c) const {
+       that_->ProduceWarning_(c);
+     }
    private:
     explicit ExecutionMessageBus(MessageWorker* that) : that_(that) {}
     MessageWorker* const that_;
   };
 
   virtual void Execute(const ExecutionMessageBus&) = 0;
-  virtual void HandleMessageCallback(RdKafka::Message*) = 0;
+  virtual void HandleMessageCallback(RdKafka::Message*, RdKafka::ErrorCode) = 0;
 
   virtual void Destroy() {
     uv_close(reinterpret_cast<uv_handle_t*>(m_async), AsyncClose_);
@@ -135,6 +144,13 @@ class MessageWorker : public ErrorAwareWorker {
     m_asyncdata.push_back(m);
     uv_async_send(m_async);
   }
+
+  void ProduceWarning_(RdKafka::ErrorCode c) {
+    scoped_mutex_lock lock(m_async_lock);
+    m_asyncwarning.push_back(c);
+    uv_async_send(m_async);
+  }
+
   NAN_INLINE static NAUV_WORK_CB(m_async_message) {
     MessageWorker *worker = static_cast<MessageWorker*>(async->data);
     worker->WorkMessage();
@@ -149,6 +165,7 @@ class MessageWorker : public ErrorAwareWorker {
   uv_async_t *m_async;
   uv_mutex_t m_async_lock;
   std::vector<RdKafka::Message*> m_asyncdata;
+  std::vector<RdKafka::ErrorCode> m_asyncwarning;
 };
 
 namespace Handle {
@@ -283,7 +300,7 @@ class KafkaConsumerConsumeLoop : public MessageWorker {
   void Execute(const ExecutionMessageBus&);
   void HandleOKCallback();
   void HandleErrorCallback();
-  void HandleMessageCallback(RdKafka::Message*);
+  void HandleMessageCallback(RdKafka::Message*, RdKafka::ErrorCode);
  private:
   NodeKafka::KafkaConsumer * consumer;
   const int m_timeout_ms;