RGW | Bucket Notification: Fix the issue of miscounting the entries to remove

Signed-off-by: Ali Masarwa <ali.saed.masarwa@gmail.com>

diff --git a/src/rgw/driver/rados/rgw_notify.cc b/src/rgw/driver/rados/rgw_notify.cc
index 969b61ff1b260c39851cb0398104430bee16a760..1d9f69498a018d07960f0c0b37e0ff1d3a123d45 100644 (file)
--- a/src/rgw/driver/rados/rgw_notify.cc
+++ b/src/rgw/driver/rados/rgw_notify.cc
@@ -350,7 +350,6 @@ private:
       is_idle = false;
       auto has_error = false;
       auto remove_entries = false;
-      uint64_t entries_to_remove = 0;
       auto entry_idx = 1U;
       tokens_waiter waiter(io_context);
       for (auto& entry : entries) {
@@ -360,8 +359,8 @@ private:
         }
 
         entries_persistency_tracker& notifs_persistency_tracker = topics_persistency_tracker[queue_name];
-        spawn::spawn(yield, [this, &notifs_persistency_tracker, &queue_name, entry_idx, total_entries, &end_marker, &remove_entries,
-                             &entries_to_remove, &has_error, &waiter, &entry](yield_context yield) {
+        spawn::spawn(yield, [this, &notifs_persistency_tracker, &queue_name, entry_idx, total_entries, &end_marker,
+                             &remove_entries, &has_error, &waiter, &entry](yield_context yield) {
             const auto token = waiter.make_token();
             auto& persistency_tracker = notifs_persistency_tracker[entry.marker];
             auto result = process_entry(this->get_cct()->_conf, persistency_tracker, entry, yield);
@@ -370,7 +369,6 @@ private:
                 entry.marker << " (" << entry_idx << "/" << total_entries << ") from: " << queue_name
                 << (result == EntryProcessingResult::Successful? " ok": " expired") << dendl;
               remove_entries = true;
-              ++entries_to_remove;
               notifs_persistency_tracker.erase(entry.marker);
             }  else {
               if (set_min_marker(end_marker, entry.marker) < 0) {
@@ -391,6 +389,14 @@ private:
 
       // delete all published entries from queue
       if (remove_entries) {
+        uint64_t entries_to_remove = 0;
+        for (const auto& entry: entries) {
+          if (end_marker == entry.marker) {
+            break;
+          }
+          entries_to_remove++;
+        }
+
         librados::ObjectWriteOperation op;
         op.assert_exists();
         rados::cls::lock::assert_locked(&op, queue_name+"_lock", 

diff --git a/src/test/rgw/bucket_notification/test_bn.py b/src/test/rgw/bucket_notification/test_bn.py
index c950400c4117849c567988ebd3103960e52c89f0..b7fe128cf2dfb4b0a1b230abfb3d278cc4d62d9f 100644 (file)
--- a/src/test/rgw/bucket_notification/test_bn.py
+++ b/src/test/rgw/bucket_notification/test_bn.py
@@ -3059,7 +3059,7 @@ def test_ps_s3_persistent_topic_stats():
     response, status = s3_notification_conf.set_config()
     assert_equal(status/100, 2)
 
-    delay = 20
+    delay = 30
     time.sleep(delay)
     http_server.close()
 
@@ -3070,7 +3070,7 @@ def test_ps_s3_persistent_topic_stats():
     assert_equal(result[1], 0)
 
     # create objects in the bucket (async)
-    number_of_objects = 10
+    number_of_objects = 100
     client_threads = []
     start_time = time.time()
     for i in range(number_of_objects):