Add a Consumer::commit overload bounded by a timeout

include/cppkafka/consumer.h

@@ -260,6 +260,48 @@ class CPPKAFKA_API Consumer : public KafkaHandleBase {
      */
     void async_commit(const TopicPartitionList& topic_partitions);
 
+    /**
+     * \brief Commits the current partition assignment, giving up after a timeout
+     *
+     * This translates into a call to rd_kafka_commit_queue on a queue private to this call.
+     * Consumer::commit calls rd_kafka_commit, which takes no timeout and waits on its reply
+     * queue until an op arrives: when no broker is reachable to send the OffsetCommit request
+     * to, librdkafka enqueues no reply and that call does not return.
+     *
+     * \param timeout How long to wait for the commit result
+     *
+     * \remark Throws HandleException carrying RD_KAFKA_RESP_ERR__TIMED_OUT once the timeout
+     * passes. The commit may still complete on the broker after that, which is safe: an offset
+     * commit is idempotent
+     *
+     * \remark Reports per-partition errors the same way the blocking commits do, through
+     * Event::get_partitions_error: the reply carries them separately from its request-level error
+     */
+    void commit(std::chrono::milliseconds timeout);
+
+    /**
+     * \brief Commits the offsets on the given topic/partitions, giving up after a timeout
+     *
+     * See commit(std::chrono::milliseconds) for how this differs from Consumer::commit and for
+     * what happens on timeout
+     *
+     * \param topic_partitions The topic/partition list to be committed
+     * \param timeout How long to wait for the commit result
+     */
+    void commit(const TopicPartitionList& topic_partitions, std::chrono::milliseconds timeout);
+
+    /**
+     * \brief Commits the offset of the given message, giving up after a timeout
+     *
+     * This is the message form of commit(std::chrono::milliseconds): the committed offset is the
+     * message's own, which is its offset + 1, the way Consumer::commit(const Message&) commits it.
+     * See commit(std::chrono::milliseconds) for how the deadline and per-partition errors behave.
+     *
+     * \param msg The message whose offset is to be committed
+     * \param timeout How long to wait for the commit result
+     */
+    void commit(const Message& msg, std::chrono::milliseconds timeout);
+
     /**
      * \brief Gets the minimum and maximum offsets for the given topic/partition
      *
@@ -511,6 +553,7 @@ class CPPKAFKA_API Consumer : public KafkaHandleBase {
     void close();
     void commit(const Message& msg, bool async);
     void commit(const TopicPartitionList* topic_partitions, bool async);
+    void commit(const TopicPartitionList* topic_partitions, std::chrono::milliseconds timeout);
     void handle_rebalance(rd_kafka_resp_err_t err, TopicPartitionList& topic_partitions);
 
     AssignmentCallback assignment_callback_;

include/cppkafka/event.h

@@ -116,6 +116,15 @@ class Event {
      */
     Error get_error() const;
 
+    /**
+     * \brief Returns the first error attached to an entry of this event's topic/partition list
+     *
+     * An RD_KAFKA_EVENT_OFFSET_COMMIT reply carries the request-level error in Event::get_error and
+     * a per-partition error on each entry of the list. Both have to be checked: a commit the broker
+     * rejected for one partition arrives with no request-level error.
+     */
+    Error get_partitions_error() const;
+
     /**
      * Gets the opaque pointer in this event
      */

src/consumer.cpp

@@ -158,6 +158,19 @@ void Consumer::async_commit(const TopicPartitionList& topic_partitions) {
     commit(&topic_partitions, true);
 }
 
+void Consumer::commit(milliseconds timeout) {
+    commit(nullptr, timeout);
+}
+
+void Consumer::commit(const TopicPartitionList& topic_partitions, milliseconds timeout) {
+    commit(&topic_partitions, timeout);
+}
+
+void Consumer::commit(const Message& msg, milliseconds timeout) {
+    TopicPartitionList offsets{TopicPartition(msg.get_topic(), msg.get_partition(), msg.get_offset() + 1)};
+    commit(offsets, timeout);
+}
+
 KafkaHandleBase::OffsetTuple Consumer::get_offsets(const TopicPartition& topic_partition) const {
     int64_t low;
     int64_t high;
@@ -310,6 +323,30 @@ void Consumer::commit(const TopicPartitionList* topic_partitions, bool async) {
     }
 }
 
+void Consumer::commit(const TopicPartitionList* topic_partitions, milliseconds timeout) {
+    // Not Queue::make_queue: its refcount workaround is for handles from rd_kafka_queue_get_*
+    Queue queue(rd_kafka_queue_new(get_handle()));
+    TopicPartitionsListPtr topic_list_handle = topic_partitions == nullptr
+        ? make_handle(nullptr)
+        : convert(*topic_partitions);
+    rd_kafka_resp_err_t error = rd_kafka_commit_queue(get_handle(), topic_list_handle.get(),
+                                                     queue.get_handle(), nullptr, nullptr);
+    check_error(error, topic_list_handle.get());
+    const Event event = queue.next_event(timeout);
+    if (!event) {
+        throw HandleException(RD_KAFKA_RESP_ERR__TIMED_OUT);
+    }
+    const Error commit_error = event.get_error();
+    if (commit_error) {
+        throw HandleException(commit_error);
+    }
+    // Same contract as the check_error(error, list) the blocking commits use
+    const Error partitions_error = event.get_partitions_error();
+    if (partitions_error) {
+        throw HandleException(partitions_error);
+    }
+}
+
 void Consumer::handle_rebalance(rd_kafka_resp_err_t error,
                                 TopicPartitionList& topic_partitions) {
     if (error == RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS) {

src/event.cpp

@@ -69,6 +69,18 @@ Error Event::get_error() const {
     return rd_kafka_event_error(handle_.get());
 }
 
+Error Event::get_partitions_error() const {
+    const rd_kafka_topic_partition_list_t* list = rd_kafka_event_topic_partition_list(handle_.get());
+    if (list != nullptr) {
+        for (int i = 0; i < list->cnt; ++i) {
+            if (list->elems[i].err != RD_KAFKA_RESP_ERR_NO_ERROR) {
+                return Error(list->elems[i].err);
+            }
+        }
+    }
+    return Error(RD_KAFKA_RESP_ERR_NO_ERROR);
+}
+
 void* Event::get_opaque() const {
     return rd_kafka_event_opaque(handle_.get());
 }

tests/CMakeLists.txt

@@ -38,6 +38,7 @@ add_executable(cppkafka_tests
     kafka_handle_base_test.cpp
     producer_test.cpp
     consumer_test.cpp
+    commit_timeout_test.cpp
     roundrobin_poll_test.cpp
     headers_test.cpp
     test_utils.cpp

tests/commit_timeout_test.cpp

@@ -0,0 +1,82 @@
+#include <chrono>
+#include <catch.hpp>
+#include "cppkafka/consumer.h"
+#include "cppkafka/exceptions.h"
+
+using std::chrono::duration_cast;
+using std::chrono::milliseconds;
+using std::chrono::steady_clock;
+
+using namespace cppkafka;
+
+namespace {
+
+// Needs no KAFKA_TEST_INSTANCE: nothing accepts connections on port 1, so the group never gets a
+// coordinator, librdkafka enqueues no commit reply, and the deadline is the only thing that can end
+// the call. The blocking overloads instead return `_WAIT_COORD` one session.timeout.ms later.
+Configuration make_config() {
+    Configuration config;
+    config.set("metadata.broker.list", "127.0.0.1:1");
+    config.set("group.id", "cppkafka_commit_timeout_test");
+    config.set("enable.auto.commit", "false");
+    config.set("enable.auto.offset.store", "false");
+    config.set("socket.timeout.ms", "1000");
+    return config;
+}
+
+template <typename Commit>
+void check_the_deadline_returns(const Commit& commit) {
+    const auto start = steady_clock::now();
+    try {
+        commit();
+        FAIL("the commit returned instead of reaching the deadline");
+    }
+    catch (const HandleException& ex) {
+        // Any other code means something answered first, and the elapsed-time bound below would then
+        // hold without the deadline being what returned.
+        CHECK(ex.get_error().get_error() == RD_KAFKA_RESP_ERR__TIMED_OUT);
+    }
+    CHECK(duration_cast<milliseconds>(steady_clock::now() - start).count() < 15000);
+}
+
+} // namespace
+
+TEST_CASE("commit bounded by a timeout", "[consumer][commit_timeout]") {
+    const TopicPartitionList offsets{TopicPartition("cppkafka_commit_timeout_test", 0, 42)};
+
+    SECTION("current assignment") {
+        Consumer consumer(make_config());
+        // A commit carrying no offset is answered `_NO_OFFSET` before the queue is polled, so a
+        // consumer holding no stored offset cannot reach the deadline.
+        consumer.assign(offsets);
+        consumer.store_offsets(offsets);
+
+        check_the_deadline_returns([&] { consumer.commit(milliseconds(3000)); });
+    }
+
+    SECTION("explicit offsets") {
+        Consumer consumer(make_config());
+
+        check_the_deadline_returns([&] { consumer.commit(offsets, milliseconds(3000)); });
+    }
+
+    SECTION("consumed message") {
+        Consumer consumer(make_config());
+
+        // rd_kafka_topic_new is the deprecated per-topic-handle API; it is here only so the
+        // fabricated message has a topic name to commit — no broker is contacted.
+        rd_kafka_topic_t * topic = rd_kafka_topic_new(consumer.get_handle(),
+                                                      "cppkafka_commit_timeout_test", nullptr);
+        REQUIRE(topic != nullptr);
+        rd_kafka_message_t bare_handle{};
+        bare_handle.err = RD_KAFKA_RESP_ERR_NO_ERROR;
+        bare_handle.rkt = topic;
+        bare_handle.partition = 0;
+        bare_handle.offset = 42;
+        Message message = Message::make_non_owning(&bare_handle);
+
+        check_the_deadline_returns([&] { consumer.commit(message, milliseconds(3000)); });
+
+        rd_kafka_topic_destroy(topic);
+    }
+}