Add error notification interface for Kafka clients

Author: kenneth-jia

include/kafka/Error.h

@@ -34,9 +34,12 @@ class Error
     // The error with rich info
     explicit Error(rd_kafka_error_t* error = nullptr): _rkError(error, RkErrorDeleter) {}
     // The error with brief info
-    explicit Error(rd_kafka_resp_err_t respErr): _rkRespErr(respErr) {}
+    explicit Error(rd_kafka_resp_err_t respErr): _respErr(respErr) {}
     // The error with detailed message
-    Error(rd_kafka_resp_err_t respErr, std::string message): _rkRespErr(respErr), _message(std::move(message)) {}
+    Error(rd_kafka_resp_err_t respErr, std::string message)
+        : _respErr(respErr), _message(std::move(message)) {}
+    Error(rd_kafka_resp_err_t respErr, std::string message, bool fatal)
+        : _respErr(respErr), _message(std::move(message)), _fatal(fatal) {}
 
     explicit operator bool() const { return static_cast<bool>(value()); }
 
@@ -58,7 +61,7 @@ class Error
       */
     int             value()        const
     {
-        return static_cast<int>(_rkError ? rd_kafka_error_code(_rkError.get()) : _rkRespErr);
+        return static_cast<int>(_rkError ? rd_kafka_error_code(_rkError.get()) : _respErr);
     }
 
     /**
@@ -67,15 +70,15 @@ class Error
     std::string     message()     const
     {
         return _message ? *_message :
-                 (_rkError ? rd_kafka_error_string(_rkError.get()) : rd_kafka_err2str(_rkRespErr));
+                 (_rkError ? rd_kafka_error_string(_rkError.get()) : rd_kafka_err2str(_respErr));
     }
 
     /**
      * Fatal error indicates that the client instance is no longer usable.
      */
     Optional<bool>  isFatal()     const
     {
-        return  _rkError ? rd_kafka_error_is_fatal(_rkError.get()) : Optional<bool>{};
+        return  _rkError ? rd_kafka_error_is_fatal(_rkError.get()) : _fatal;
     }
 
     /**
@@ -87,9 +90,10 @@ class Error
     }
 
 private:
-    rd_kafka_error_shared_ptr _rkError;       // For error with rich info
-    rd_kafka_resp_err_t       _rkRespErr{};   // For error with a simple response code
-    Optional<std::string>     _message;       // For additional detailed message
+    rd_kafka_error_shared_ptr _rkError;     // For error with rich info
+    rd_kafka_resp_err_t       _respErr{};   // For error with a simple response code
+    Optional<std::string>     _message;     // Additional detailed message (if any)
+    Optional<bool>            _fatal;       // Fatal flag (if any)
 };
 
 } // end of KAFKA_API

include/kafka/KafkaClient.h

@@ -32,6 +32,7 @@ class KafkaClient
 protected:
     using ConfigCallbacksRegister = std::function<void(rd_kafka_conf_t*)>;
     using StatsCallback           = std::function<void(const std::string&)>;
+    using ErrorCallback           = std::function<void(const Error&)>;
 
     enum class ClientType { KafkaConsumer, KafkaProducer, AdminClient };
     static std::string getClientTypeString(ClientType type)
@@ -91,6 +92,11 @@ class KafkaClient
      */
     void setStatsCallback(StatsCallback cb) { _statsCb = std::move(cb); }
 
+    /**
+     * Set callback for error notification.
+     */
+    void setErrorCallback(ErrorCallback cb) { _errorCb = std::move(cb); }
+
     /**
      * Return the properties which took effect.
      */
@@ -172,6 +178,9 @@ class KafkaClient
     // Statistics callback (for librdkafka)
     static int statsCallback(rd_kafka_t* rk, char* jsonStrBuf, size_t jsonStrLen, void* opaque);
 
+    // Error callback (for librdkafka)
+    static void errorCallback(rd_kafka_t* rk, int err, const char* reason, void* opaque);
+
     // Validate properties (and fix it if necesary)
     static Properties validateAndReformProperties(const Properties& origProperties);
 
@@ -191,6 +200,7 @@ class KafkaClient
     Logger              _logger;
     Properties          _properties;
     StatsCallback       _statsCb;
+    ErrorCallback       _errorCb;
     rd_kafka_unique_ptr _rk;
 
     // Log callback (for class instance)
@@ -199,6 +209,9 @@ class KafkaClient
     // Stats callback (for class instance)
     void onStats(const std::string& jsonString);
 
+    // Error callback (for class instance)
+    void onError(const Error& error);
+
     static const constexpr char* BOOTSTRAP_SERVERS = "bootstrap.servers";
     static const constexpr char* CLIENT_ID         = "client.id";
     static const constexpr char* LOG_LEVEL         = "log_level";
@@ -341,6 +354,9 @@ KafkaClient::KafkaClient(ClientType                     clientType,
     // Statistics Callback
     rd_kafka_conf_set_stats_cb(rk_conf.get(), KafkaClient::statsCallback);
 
+    // Error Callback
+    rd_kafka_conf_set_error_cb(rk_conf.get(), KafkaClient::errorCallback);
+
     // Other Callbacks
     if (registerCallbacks)
     {
@@ -466,6 +482,32 @@ KafkaClient::statsCallback(rd_kafka_t* rk, char* jsonStrBuf, size_t jsonStrLen,
     return 0;
 }
 
+inline void
+KafkaClient::onError(const Error& error)
+{
+    if (_errorCb) _errorCb(error);
+}
+
+inline void
+KafkaClient::errorCallback(rd_kafka_t* rk, int err, const char* reason, void* /*opaque*/)
+{
+    auto respErr = static_cast<rd_kafka_resp_err_t>(err);
+
+    Error error;
+    if (respErr != RD_KAFKA_RESP_ERR__FATAL)
+    {
+        error = Error{respErr, reason};
+    }
+    else
+    {
+        LogBuffer<LOG_BUFFER_SIZE> errInfo;
+        respErr = rd_kafka_fatal_error(rk, errInfo.str(), errInfo.capacity());
+        error = Error{respErr, errInfo.c_str(), true};
+    }
+
+    kafkaClient(rk).onError(error);
+}
+
 inline Optional<BrokerMetadata>
 KafkaClient::fetchBrokerMetadata(const std::string& topic, std::chrono::milliseconds timeout, bool disableErrorLogging)
 {