rgw/notification: Store the value of persistent_queue for existing persistent topic.

Fixes https://tracker.ceph.com/issues/66097

Signed-off-by: kchheda3 <[email protected]>

Author: kchheda3

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

@@ -662,7 +662,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,
@@ -674,6 +674,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
     _, result = admin(['topic', 'rm', '--topic', topic_name+'_3', '--tenant', tenant], get_config_cluster())