diff --git a/python/ray/air/tests/test_remote_storage.py b/python/ray/air/tests/test_remote_storage.py
index 07ca5ea7aecf..49c93c25e95a 100644
--- a/python/ray/air/tests/test_remote_storage.py
+++ b/python/ray/air/tests/test_remote_storage.py
@@ -1,4 +1,7 @@
 import os
+import threading
+from unittest.mock import patch
+
 import pytest
 import shutil
 import tempfile
@@ -7,6 +10,7 @@
     upload_to_uri,
     download_from_uri,
 )
+from ray.tune.utils.file_transfer import _get_recursive_files_and_stats
 
 
 @pytest.fixture
@@ -126,6 +130,33 @@ def test_upload_exclude_multimatch(temp_data_dirs):
     assert_file(False, tmp_target, "subdir_exclude/something/somewhere.txt")
 
 
+def test_get_recursive_files_race_con(temp_data_dirs):
+    tmp_source, _ = temp_data_dirs
+
+    def run(event):
+        lst = os.lstat
+
+        def waiting_lstat(*args, **kwargs):
+            event.wait()
+            return lst(*args, **kwargs)
+
+        with patch("os.lstat", wraps=waiting_lstat):
+            _get_recursive_files_and_stats(tmp_source)
+
+    event = threading.Event()
+
+    get_thread = threading.Thread(target=run, args=(event,))
+    get_thread.start()
+
+    os.remove(os.path.join(tmp_source, "level0.txt"))
+    event.set()
+
+    get_thread.join()
+
+    assert_file(False, tmp_source, "level0.txt")
+    assert_file(True, tmp_source, "level0_exclude.txt")
+
+
 if __name__ == "__main__":
     import sys
 
diff --git a/python/ray/tune/utils/file_transfer.py b/python/ray/tune/utils/file_transfer.py
index b2dd4ddc8868..f877f321294c 100644
--- a/python/ray/tune/utils/file_transfer.py
+++ b/python/ray/tune/utils/file_transfer.py
@@ -214,9 +214,14 @@ def _get_recursive_files_and_stats(path: str) -> Dict[str, Tuple[float, int]]:
     for root, dirs, files in os.walk(path, topdown=False):
         rel_root = os.path.relpath(root, path)
         for file in files:
-            key = os.path.join(rel_root, file)
-            stat = os.lstat(os.path.join(path, key))
-            files_stats[key] = stat.st_mtime, stat.st_size
+            try:
+                key = os.path.join(rel_root, file)
+                stat = os.lstat(os.path.join(path, key))
+                files_stats[key] = stat.st_mtime, stat.st_size
+            except FileNotFoundError:
+                # Race condition: If a file is deleted while executing this
+                # method, just continue and don't include the file in the stats
+                pass
 
     return files_stats