diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index 7da6696b0e2..a49f70061cf 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -3634,15 +3634,28 @@ def _execute_storage_mounts(self, handle: CloudVmRayResourceHandle,
                          if storage_obj.source else storage_obj.name)
             if isinstance(src_print, list):
                 src_print = ', '.join(src_print)
-            backend_utils.parallel_data_transfer_to_nodes(
-                runners,
-                source=src_print,
-                target=dst,
-                cmd=mount_cmd,
-                run_rsync=False,
-                action_message='Mounting',
-                log_path=log_path,
-            )
+            try:
+                backend_utils.parallel_data_transfer_to_nodes(
+                    runners,
+                    source=src_print,
+                    target=dst,
+                    cmd=mount_cmd,
+                    run_rsync=False,
+                    action_message='Mounting',
+                    log_path=log_path,
+                )
+            except exceptions.CommandError as e:
+                if e.returncode == exceptions.MOUNT_PATH_NON_EMPTY_CODE:
+                    mount_path = (f'{colorama.Fore.RED}'
+                                  f'{colorama.Style.BRIGHT}{dst}'
+                                  f'{colorama.Style.RESET_ALL}')
+                    error_msg = (f'Mount path {mount_path} is non-empty.'
+                                 f' {mount_path} may be a standard unix '
+                                 f'path or may contain files from a previous'
+                                 f' task. To fix, change the mount path'
+                                 f' to an empty or non-existent path.')
+                    raise RuntimeError(error_msg) from None
+
         end = time.time()
         logger.debug(f'Storage mount sync took {end - start} seconds.')
 
diff --git a/sky/data/mounting_utils.py b/sky/data/mounting_utils.py
index 1bde951f8b7..1f071669912 100644
--- a/sky/data/mounting_utils.py
+++ b/sky/data/mounting_utils.py
@@ -3,6 +3,8 @@
 import textwrap
 from typing import Optional
 
+from sky import exceptions
+
 
 def get_mounting_command(
     mount_path: str,
@@ -59,8 +61,8 @@ def get_mounting_command(
         else
           # Check if mount path contains files
           if [ "$(ls -A $MOUNT_PATH)" ]; then
-            echo "Mount path $MOUNT_PATH is not empty. Please make sure its empty."
-            exit 1
+            echo "Mount path $MOUNT_PATH is not empty. Please mount to another path or remove it first."
+            exit {exceptions.MOUNT_PATH_NON_EMPTY_CODE}
           fi
         fi
         echo "Mounting $SOURCE_BUCKET to $MOUNT_PATH with $MOUNT_BINARY..."
diff --git a/sky/exceptions.py b/sky/exceptions.py
index 0211cb81efa..3dfb7172d17 100644
--- a/sky/exceptions.py
+++ b/sky/exceptions.py
@@ -10,6 +10,8 @@
 KEYBOARD_INTERRUPT_CODE = 130
 SIGTSTP_CODE = 146
 RSYNC_FILE_NOT_FOUND_CODE = 23
+# Arbitrarily chosen value. Used in SkyPilot's storage mounting scripts
+MOUNT_PATH_NON_EMPTY_CODE = 42
 
 
 class ResourcesUnavailableError(Exception):