code: def on_cluster_created(self, ray_cluster_handler):
      db_api_entry = get_db_entry_point()
  
      try:
          get_databricks_function("displayHTML")(
              "<b style='background-color:yellow;'>When you are using Ray on Spark "
              "cluster, you only pay for Spark cluster usage.</b>"
          )
      except Exception:
          pass
  
      if ray_cluster_handler.autoscale or self.is_global:
          # Disable auto shutdown if
          # 1) autoscaling enabled
          #  because in autoscaling mode, background spark job will be killed
          #  automatically when ray cluster is idle.
          # 2) global mode cluster
          #  Because global mode cluster is designed to keep running until
          #  user request to shut down it, and global mode cluster is shared
          #  by other users, the code here cannot track usage from other users
          #  so that we don't know whether it is safe to shut down the global
          #  cluster automatically.
          auto_shutdown_minutes = 0
      else:
          auto_shutdown_minutes = float(
              os.environ.get(
                  DATABRICKS_RAY_ON_SPARK_AUTOSHUTDOWN_MINUTES, "30"
              )
          )
      if auto_shutdown_minutes == 0:
          _logger.info(
              "The Ray cluster will keep running until you manually detach the "
              "Databricks notebook or call "
              "`ray.util.spark.shutdown_ray_cluster()`."
          )
          return
      if auto_shutdown_minutes < 0:
          raise ValueError(
              "You must set "
              f"'{DATABRICKS_RAY_ON_SPARK_AUTOSHUTDOWN_MINUTES}' "
              "to a value >= 0."
          )
  
      try:
          db_api_entry.getIdleTimeMillisSinceLastNotebookExecution()
      except Exception:
          _logger.warning(
              "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 "
          f"{auto_shutdown_minutes} minutes. You can change the "
          "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."
      )
  
      def auto_shutdown_watcher():
          auto_shutdown_millis = auto_shutdown_minutes * 60 * 1000
          while True:
              if ray_cluster_handler.is_shutdown:
                  # The cluster is shut down. The watcher thread exits.
                  return
  
              idle_time = (
                  db_api_entry.getIdleTimeMillisSinceLastNotebookExecution()
              )
  
              if idle_time > auto_shutdown_millis:
                  from ray.util.spark import cluster_init
  
                  with cluster_init._active_ray_cluster_rwlock:
                      if (
                          ray_cluster_handler
                          is cluster_init._active_ray_cluster
                      ):
                          cluster_init.shutdown_ray_cluster()
                  return
  
              time.sleep(DATABRICKS_AUTO_SHUTDOWN_POLL_INTERVAL_SECONDS)
  
      threading.Thread(
          target=auto_shutdown_watcher, daemon=True
      ).start()
  
file_path: ray/python/ray/util/spark/databricks_hook.py
imports: import os
  from .start_hook_base import RayOnSparkStartHook
  from .utils import get_spark_session
  import logging
  import threading
  import time
label_desc: Use image node_id 3 for calling the DefaultDatabricksRayOnSparkStartHook obj's underlying member method code with example usage: obj.on_cluster_created(ray_cluster_handler) without return types