scottk/wlan-cloud-lib-cppkafka
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-cloud-lib-cppkafka.git synced 2025-11-04 04:27:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

Removed ReturnType. Throw on error from inside do_commit() as well as from perform()

Browse Source
This commit is contained in:
Alexander Damian
2019-04-24 16:42:56 -04:00
parent d84b75ca9d
commit effdf7fb95
3 changed files with 22 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
include/cppkafka/utils/backoff_committer.h
View File

@@ -137,32 +137,28 @@ public:
private:
// 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 T>
ReturnType do_commit(const T* object) {
ReturnType rt;
template <typename...Args>
bool do_commit(Args&&...args) {
try {
if (!object) {
consumer_.commit();
}
else {
consumer_.commit(*object);
}
consumer_.commit(std::forward<Args>(args)...);
// If the commit succeeds, we're done.
return true;
}
catch (const HandleException& ex) {
rt.error_ = ex.get_error();
Error error = ex.get_error();
// If there were actually no offsets to commit, return. Retrying won't solve
// anything here
if (rt.error_ != RD_KAFKA_RESP_ERR__NO_OFFSET) {
// If there's no callback or if returns true for this message, keep committing.
// Otherwise abort.
CallbackInvoker<ErrorCallback> callback("backoff committer", callback_, &consumer_);
if (!callback || callback(rt.error_)) {
rt.abort_ = false; //continue retrying
}
if (error == RD_KAFKA_RESP_ERR__NO_OFFSET) {
throw ex; //abort
}
// If there's a callback and it returns false for this message, abort.
// Otherwise keep committing.
CallbackInvoker<ErrorCallback> callback("backoff committer", callback_, &consumer_);
if (callback && !callback(error)) {
throw ex; //abort
}
}
return rt;
return false; //continue
}
Consumer& consumer_;