Merge PR #66168 into tentacle

* refs/pull/66168/head:
	rgw/multisite: check the local bucket's versioning status when replicating deletion from remote

Reviewed-by: Shilpa Jagannath <smanjara@redhat.com>

diff --cc src/test/rgw/rgw_multi/tests.py
index d98ed962d7ce3f643c4c10c416ea486deb0a40ea,050dc8d54f2b4381bace07405abe96c28682681d..8cad5fe8ecc6ac9c5378f6fe61bdee4bae035460
--- 1/src/test/rgw/rgw_multi/tests.py
--- 2/src/test/rgw/rgw_multi/tests.py
+++ b/src/test/rgw/rgw_multi/tests.py
@@@ -1270,15 -1171,39 +1270,41 @@@ def test_suspended_delete_marker_increm
  
      zonegroup_bucket_checkpoint(zonegroup_conns, bucket.name)
  
+ def test_suspended_delete_marker_incremental_sync():
+     zonegroup = realm.master_zonegroup()
+     zonegroup_conns = ZonegroupConns(zonegroup)
+     zone = zonegroup_conns.rw_zones[0]
+ 
+     # create a versioned bucket
+     bucket = zone.create_bucket(gen_bucket_name())
+     log.debug('created bucket=%s', bucket.name)
+     bucket.configure_versioning(True)
+     bucket.configure_versioning(False)
+ 
+     zonegroup_meta_checkpoint(zonegroup)
+ 
+     obj = 'obj'
+ 
+     # upload a dummy object and wait for sync. this forces each zone to finish
+     # a full sync and switch to incremental
+     new_key(zone, bucket, obj).set_contents_from_string('')
+     zonegroup_bucket_checkpoint(zonegroup_conns, bucket.name)
+ 
+     # create several a delete marker on source zone and let it sync
+     key = new_key(zone, bucket, obj)
+     key.delete()
+ 
+     zonegroup_bucket_checkpoint(zonegroup_conns, bucket.name)
+ 
  def test_bucket_versioning():
      buckets, zone_bucket = create_bucket_per_zone_in_realm()
 -    for _, bucket in zone_bucket:
 -        bucket.configure_versioning(True)
 -        res = bucket.get_versioning_status()
 -        key = 'Versioning'
 -        assert(key in res and res[key] == 'Enabled')
 +    for zone, bucket in zone_bucket:
 +        zone.s3_client.put_bucket_versioning(
 +            Bucket=bucket.name,
 +            VersioningConfiguration={'Status': 'Enabled'}
 +        )
 +        response = zone.s3_client.get_bucket_versioning(Bucket=bucket.name)
 +        assert response.get('Status') == 'Enabled'
  
  def test_bucket_acl():
      buckets, zone_bucket = create_bucket_per_zone_in_realm()