[spark] Refine some text in Ray on Spark exception messages and warning messages (ray-project#32162)

See follow-up comments in ray-project#31962

Signed-off-by: Weichen Xu <weichen.xu@databricks.com>

Author: WeichenXu123

python/ray/util/spark/cluster_init.py

@@ -127,7 +127,7 @@ def wait_until_ready(self):
                 # in this case, raise error directly.
                 if self.background_job_exception is not None:
                     raise RuntimeError(
-                        "Ray workers have exited."
+                        "Ray workers failed to start."
                     ) from self.background_job_exception
 
                 cur_alive_worker_count = (
@@ -406,6 +406,8 @@ def _setup_ray_cluster(
     ray_head_ip = socket.gethostbyname(get_spark_application_driver_host(spark))
     ray_head_port = get_random_unused_port(ray_head_ip, min_port=9000, max_port=10000)
 
+    # Make a copy for head_node_options to avoid changing original dict in user code.
+    head_node_options = head_node_options.copy()
     include_dashboard = head_node_options.pop("include_dashboard", None)
     ray_dashboard_port = head_node_options.pop("dashboard_port", None)
 
@@ -851,7 +853,7 @@ def setup_ray_cluster(
             collect their logs to the specified path. On Databricks Runtime, we
             recommend you to specify a local path starts with '/dbfs/', because the
             path mounts with a centralized storage device and stored data is persisted
-            after databricks spark cluster terminated.
+            after Databricks spark cluster terminated.
 
     Returns:
         The address of the initiated Ray cluster on spark.

python/ray/util/spark/databricks_hook.py

@@ -93,7 +93,7 @@ def on_cluster_created(self, ray_cluster_handler):
             _logger.warning(
                 "Registering Ray cluster spark job as background job failed. "
                 "You need to manually call `ray.util.spark.shutdown_ray_cluster()` "
-                "before detaching your databricks notebook."
+                "before detaching your Databricks notebook."
             )
 
         auto_shutdown_minutes = float(
@@ -102,7 +102,7 @@ def on_cluster_created(self, ray_cluster_handler):
         if auto_shutdown_minutes == 0:
             _logger.info(
                 "The Ray cluster will keep running until you manually detach the "
-                "databricks notebook or call "
+                "Databricks notebook or call "
                 "`ray.util.spark.shutdown_ray_cluster()`."
             )
             return
@@ -117,26 +117,24 @@ def on_cluster_created(self, ray_cluster_handler):
             db_api_entry.getIdleTimeMillisSinceLastNotebookExecution()
         except Exception:
             _logger.warning(
-                "Databricks `getIdleTimeMillisSinceLastNotebookExecution` API "
-                "is unavailable, it is probably because that "
-                "your current Databricks Runtime version does not support API "
-                "`getIdleTimeMillisSinceLastNotebookExecution`, we cannot "
-                "automatically shut down Ray cluster when databricks notebook "
-                "is inactive, you need to manually detach databricks notebook "
+                "Failed to retrieve idle time since last notebook execution, "
+                "so that we cannot automatically shut down Ray cluster when "
+                "Databricks notebook is inactive for the specified minutes. "
+                "You need to manually detach Databricks notebook "
                 "or call `ray.util.spark.shutdown_ray_cluster()` to shut down "
                 "Ray cluster on spark."
             )
             return
 
         _logger.info(
             "The Ray cluster will be shut down automatically if you don't run "
-            "commands on the databricks notebook for "
+            "commands on the Databricks notebook for "
             f"{auto_shutdown_minutes} minutes. You can change the "
-            "automatically shutdown minutes by setting "
+            "auto-shutdown minutes by setting "
             f"'{DATABRICKS_RAY_ON_SPARK_AUTOSHUTDOWN_MINUTES}' environment "
             "variable, setting it to 0 means that the Ray cluster keeps running "
             "until you manually call `ray.util.spark.shutdown_ray_cluster()` or "
-            "detach databricks notebook."
+            "detach Databricks notebook."
         )
 
         def auto_shutdown_watcher():