Merge pull request mfontanini#187 from accelerated/master

Added commit termination callback functionality in BackoffComitter

Author: mfontanini

include/cppkafka/error.h

@@ -42,6 +42,10 @@ namespace cppkafka {
  */
 class CPPKAFKA_API Error {
 public:
+    /**
+     * @brief Constructs an error object with RD_KAFKA_RESP_ERR_NO_ERROR
+     */
+    Error() = default;
     /**
      * Constructs an error object
      */
@@ -77,7 +81,7 @@ class CPPKAFKA_API Error {
      */
     CPPKAFKA_API friend std::ostream& operator<<(std::ostream& output, const Error& rhs);
 private:
-    rd_kafka_resp_err_t error_;
+    rd_kafka_resp_err_t error_{RD_KAFKA_RESP_ERR_NO_ERROR};
 };
 
 } // cppkafka

include/cppkafka/exceptions.h

@@ -134,6 +134,14 @@ class CPPKAFKA_API QueueException : public Exception {
     Error error_;
 };
 
+/**
+ * Backoff performer has no more retries left for a specific action.
+ */
+class CPPKAFKA_API ActionTerminatedException : public Exception {
+public:
+    ActionTerminatedException(const std::string& error);
+};
+
 } // cppkafka
 
 #endif // CPPKAFKA_EXCEPTIONS_H

include/cppkafka/utils/backoff_committer.h

@@ -100,10 +100,18 @@ class CPPKAFKA_API BackoffCommitter : public BackoffPerformer {
      */
     void set_error_callback(ErrorCallback callback);
 
+    /**
+     * \brief Commits the current partition assignment synchronously
+     *
+     * This will call Consumer::commit() until either the message is successfully
+     * committed or the error callback returns false (if any is set).
+     */
+    void commit();
+    
     /**
      * \brief Commits the given message synchronously
      *
-     * This will call Consumer::commit until either the message is successfully
+     * This will call Consumer::commit(msg) until either the message is successfully
      * committed or the error callback returns false (if any is set).
      *
      * \param msg The message to be committed
@@ -113,7 +121,7 @@ class CPPKAFKA_API BackoffCommitter : public BackoffPerformer {
     /**
      * \brief Commits the offsets on the given topic/partitions synchronously
      *
-     * This will call Consumer::commit until either the offsets are successfully
+     * This will call Consumer::commit(topic_partitions) until either the offsets are successfully
      * committed or the error callback returns false (if any is set).
      *
      * \param topic_partitions The topic/partition list to be committed
@@ -127,25 +135,30 @@ class CPPKAFKA_API BackoffCommitter : public BackoffPerformer {
      */
     Consumer& get_consumer();
 private:
-    // Return true to abort and false to continue committing
-    template <typename T>
-    bool do_commit(const T& object) {
+    // If the ReturnType contains 'true', we abort committing. Otherwise we continue.
+    // The second member of the ReturnType contains the RdKafka error if any.
+    template <typename...Args>
+    bool do_commit(Args&&...args) {
         try {
-            consumer_.commit(object);
-            // If the commit succeeds, we're done
+            consumer_.commit(std::forward<Args>(args)...);
+            // If the commit succeeds, we're done.
             return true;
         }
         catch (const HandleException& ex) {
+            Error error = ex.get_error();
             // If there were actually no offsets to commit, return. Retrying won't solve
-            // anything here
-            if (ex.get_error() == RD_KAFKA_RESP_ERR__NO_OFFSET) {
-                return true;
+            // anything here.
+            if (error == RD_KAFKA_RESP_ERR__NO_OFFSET) {
+                return true; //not considered an error.
             }
             // If there's a callback and it returns false for this message, abort.
             // Otherwise keep committing.
             CallbackInvoker<ErrorCallback> callback("backoff committer", callback_, &consumer_);
-            return callback && !callback(ex.get_error());
+            if (callback && !callback(error)) {
+                throw ex; //abort
+            }
         }
+        return false; //continue
     }
 
     Consumer& consumer_;

include/cppkafka/utils/backoff_performer.h

@@ -34,6 +34,7 @@
 #include <functional>
 #include <thread>
 #include "../consumer.h"
+#include "../exceptions.h"
 
 namespace cppkafka {
 
@@ -123,7 +124,7 @@ class CPPKAFKA_API BackoffPerformer {
             auto start = std::chrono::steady_clock::now();
             // If the callback returns true, we're done
             if (callback()) {
-                return;
+                return; //success
             }
             auto end = std::chrono::steady_clock::now();
             auto time_elapsed = end - start;
@@ -134,6 +135,8 @@ class CPPKAFKA_API BackoffPerformer {
             // Increase out backoff depending on the policy being used
             backoff = increase_backoff(backoff);
         }
+        // No more retries left or we have a terminal error.
+        throw ActionTerminatedException("Commit failed: no more retries.");
     }
 private:
     TimeUnit increase_backoff(TimeUnit backoff);

src/utils/backoff_committer.cpp

@@ -43,14 +43,20 @@ void BackoffCommitter::set_error_callback(ErrorCallback callback) {
     callback_ = move(callback);
 }
 
+void BackoffCommitter::commit() {
+    perform([&] {
+        return do_commit();
+    });
+}
+
 void BackoffCommitter::commit(const Message& msg) {
-    perform([&] { 
+    perform([&] {
         return do_commit(msg);
     });
 }
 
 void BackoffCommitter::commit(const TopicPartitionList& topic_partitions) {
-    perform([&] { 
+    perform([&] {
         return do_commit(topic_partitions);
     });
 }