Merge pull request ceph#57536 from kchheda3/wip-fix-persistent-queue-regression

 rgw/notification: Store the value of `persistent_queue` for existing topics and continue commiting events for all topics subscribed to given bucket

Reviewed-by: Yuval Lifshitz <[email protected]>

Author: cbodley

src/rgw/driver/rados/rgw_notify.cc

@@ -1111,7 +1111,7 @@ int publish_reserve(const DoutPrefixProvider* dpp,
           // either the topic is deleted but the corresponding notification
           // still exist or in v2 mode the notification could have synced first
           // but topic is not synced yet.
-          return 0;
+          continue;
         }
         ldpp_dout(res.dpp, 1)
             << "WARN: Using the stored topic from bucket notification struct."

src/rgw/rgw_rest_pubsub.cc

@@ -422,6 +422,8 @@ void RGWPSCreateTopicOp::execute(optional_yield y) {
                          << op_ret << dendl;
       return;
     }
+  } else if (already_persistent) {  // redundant call to CreateTopic
+    dest.persistent_queue = topic->dest.persistent_queue;
   }
   const RGWPubSub ps(driver, get_account_or_tenant(s->owner.id), *s->penv.site);
   op_ret = ps.create_topic(this, topic_name, dest, topic_arn.to_string(),
@@ -882,7 +884,7 @@ void RGWPSSetTopicAttributesOp::execute(optional_yield y) {
           << op_ret << dendl;
       return;
     }
-  } else if (already_persistent) {
+  } else if (already_persistent && !topic_needs_queue(dest)) {
     // changing the persistent topic to non-persistent.
     op_ret = driver->remove_persistent_topic(this, y, result.dest.persistent_queue);
     if (op_ret != -ENOENT && op_ret < 0) {

src/test/rgw/bucket_notification/test_bn.py

@@ -759,7 +759,7 @@ def test_ps_s3_topic_admin_on_master():
     assert_equal(topic_arn2,
                  'arn:aws:sns:' + zonegroup + ':' + tenant + ':' + topic_name + '_2')
     endpoint_address = 'http://127.0.0.1:9002'
-    endpoint_args = 'push-endpoint='+endpoint_address
+    endpoint_args = 'push-endpoint=' + endpoint_address + '&persistent=true'
     topic_conf3 = PSTopicS3(conn, topic_name+'_3', zonegroup, endpoint_args=endpoint_args)
     topic_arn3 = topic_conf3.set_config()
     assert_equal(topic_arn3,
@@ -770,6 +770,24 @@ def test_ps_s3_topic_admin_on_master():
     assert_equal(parsed_result['arn'], topic_arn3)
     matches = [tenant, UID_PREFIX]
     assert_true( all([x in parsed_result['owner'] for x in matches]))
+    assert_equal(parsed_result['dest']['persistent_queue'],
+                 tenant + ":" + topic_name + '_3')
+
+    # recall CreateTopic and verify the owner and persistent_queue remain same.
+    topic_conf3 = PSTopicS3(conn, topic_name + '_3', zonegroup,
+                            endpoint_args=endpoint_args)
+    topic_arn3 = topic_conf3.set_config()
+    assert_equal(topic_arn3,
+                 'arn:aws:sns:' + zonegroup + ':' + tenant + ':' + topic_name + '_3')
+    # get topic 3 via commandline
+    result = admin(
+      ['topic', 'get', '--topic', topic_name + '_3', '--tenant', tenant],
+      get_config_cluster())
+    parsed_result = json.loads(result[0])
+    assert_equal(parsed_result['arn'], topic_arn3)
+    assert_true(all([x in parsed_result['owner'] for x in matches]))
+    assert_equal(parsed_result['dest']['persistent_queue'],
+                 tenant + ":" + topic_name + '_3')
 
     # delete topic 3
     remove_topic(topic_name + '_3', tenant)