rgw/pubsub: prevent kafka thread from spinning when there are no

messages

Signed-off-by: Yuval Lifshitz <yuvalif@yahoo.com>
(cherry picked from commit 9cce2381cbeba3409b8d79659a6ba502985afacb)
Signed-off-by: Yuval Lifshitz <yuvalif@yahoo.com>
Fixes: https://tracker.ceph.com/issues/43156
Original Fixes: https://tracker.ceph.com/issues/43104

diff --git a/src/rgw/rgw_kafka.cc b/src/rgw/rgw_kafka.cc
index 0d247f16d3324ced6e28b29acdfb3cfe8ff4ff52..dfaefdfb270710f9165b23f7fe872f30398b131d 100644 (file)
--- a/src/rgw/rgw_kafka.cc
+++ b/src/rgw/rgw_kafka.cc
@@ -399,9 +399,9 @@ private:
     while (!stopped) {
 
       // publish all messages in the queue
-      auto event_count = 0U;
-      const auto count = messages.consume_all(std::bind(&Manager::publish_internal, this, std::placeholders::_1));
-      dequeued += count;
+      auto reply_count = 0U;
+      const auto send_count = messages.consume_all(std::bind(&Manager::publish_internal, this, std::placeholders::_1));
+      dequeued += send_count;
       ConnectionList::iterator conn_it;
       ConnectionList::const_iterator end_it;
       {
@@ -440,14 +440,14 @@ private:
           INCREMENT_AND_CONTINUE(conn_it);
         }
 
-        event_count += rd_kafka_poll(conn->producer, read_timeout_ms);
+        reply_count += rd_kafka_poll(conn->producer, read_timeout_ms);
 
         // just increment the iterator
         ++conn_it;
       }
       // if no messages were received or published
       // across all connection, sleep for 100ms
-      if (count == 0 && event_count) {
+      if (send_count == 0 && reply_count == 0) {
         std::this_thread::sleep_for(std::chrono::milliseconds(100));
       }
     }