diff --git a/qa/tasks/cephfs/mount.py b/qa/tasks/cephfs/mount.py
index e9341141f63..e301f802250 100644
--- a/qa/tasks/cephfs/mount.py
+++ b/qa/tasks/cephfs/mount.py
@@ -1189,7 +1189,8 @@ class CephFSMount(object):
         self.background_procs.append(rproc)
         return rproc
 
-    def create_n_files(self, fs_path, count, sync=False, dirsync=False, unlink=False, finaldirsync=False):
+    def create_n_files(self, fs_path, count, sync=False, dirsync=False,
+                       unlink=False, finaldirsync=False, hard_links=0):
         """
         Create n files.
 
@@ -1197,6 +1198,7 @@ class CephFSMount(object):
         :param dirsync: sync the containing directory after closing the file
         :param unlink: unlink the file after closing
         :param finaldirsync: sync the containing directory after closing the last file
+        :param hard_links: create given number of hard link(s) for each file
         """
 
         assert(self.is_mounted())
@@ -1205,8 +1207,12 @@ class CephFSMount(object):
 
         pyscript = dedent(f"""
             import os
+            import uuid
 
             n = {count}
+            create_hard_links = False
+            if {hard_links} > 0:
+                create_hard_links = True
             path = "{abs_path}"
 
             dpath = os.path.dirname(path)
@@ -1227,6 +1233,9 @@ class CephFSMount(object):
                         os.unlink(fpath)
                     if {dirsync}:
                         os.fsync(dirfd)
+                    if create_hard_links:
+                        for j in range({hard_links}):
+                            os.system(f"ln {{fpath}} {{dpath}}/{{fnameprefix}}_{{i}}_{{uuid.uuid4()}}")     
                 if {finaldirsync}:
                     os.fsync(dirfd)
             finally:
diff --git a/qa/tasks/cephfs/test_scrub_checks.py b/qa/tasks/cephfs/test_scrub_checks.py
index 0e84f7ed264..f88929ad265 100644
--- a/qa/tasks/cephfs/test_scrub_checks.py
+++ b/qa/tasks/cephfs/test_scrub_checks.py
@@ -296,6 +296,36 @@ class TestScrubChecks(CephFSTestCase):
         command = "flush_path /"
         self.asok_command(mds_rank, command, success_validator)
 
+    def scrub_with_stray_evaluation(self, fs, mnt, path, flag, files=2000,
+                                    _hard_links=3):
+        fs.set_allow_new_snaps(True)
+
+        test_dir = "stray_eval_dir"
+        mnt.run_shell(["mkdir", test_dir])
+        client_path = os.path.join(mnt.mountpoint, test_dir)
+        mnt.create_n_files(fs_path=f"{test_dir}/file", count=files,
+                           hard_links=_hard_links)
+        mnt.run_shell(["mkdir", f"{client_path}/.snap/snap1-{test_dir}"])
+        mnt.run_shell(f"find {client_path}/ -type f -delete")
+        mnt.run_shell(["rmdir", f"{client_path}/.snap/snap1-{test_dir}"])
+        perf_dump = fs.rank_tell(["perf", "dump"], 0)
+        self.assertNotEqual(perf_dump.get('mds_cache').get('num_strays'),
+                            0, "mdcache.num_strays is zero")
+
+        log.info(
+            f"num of strays: {perf_dump.get('mds_cache').get('num_strays')}")
+
+        out_json = fs.run_scrub(["start", path, flag])
+        self.assertNotEqual(out_json, None)
+        self.assertEqual(out_json["return_code"], 0)
+
+        self.assertEqual(
+            fs.wait_until_scrub_complete(tag=out_json["scrub_tag"]), True)
+
+        perf_dump = fs.rank_tell(["perf", "dump"], 0)
+        self.assertEqual(int(perf_dump.get('mds_cache').get('num_strays')),
+                         0, "mdcache.num_strays is non-zero")
+
     def test_scrub_repair(self):
         mds_rank = 0
         test_dir = "scrub_repair_path"
@@ -332,6 +362,20 @@ class TestScrubChecks(CephFSTestCase):
         # fragstat should be fixed
         self.mount_a.run_shell(["rmdir", test_dir])
 
+    def test_stray_evaluation_with_scrub(self):
+        """
+        test that scrub can iterate over ~mdsdir and evaluate strays
+        """
+        self.scrub_with_stray_evaluation(self.fs, self.mount_a, "~mdsdir",
+                                         "recursive")
+
+    def test_flag_scrub_mdsdir(self):
+        """
+        test flag scrub_mdsdir
+        """
+        self.scrub_with_stray_evaluation(self.fs, self.mount_a, "/",
+                                         "scrub_mdsdir")
+
     @staticmethod
     def json_validator(json_out, rc, element, expected_value):
         if rc != 0: