Merge branch 'master' into dag-max-buffered-queue

Signed-off-by: dayshah <dhyey2019@gmail.com>

python/ray/dag/compiled_dag_node.py

@@ -2051,7 +2051,15 @@ def _execute_until(
 
             # Fetch results from each output channel up to execution_index and cache
             # them separately to enable individual retrieval
-            result = self._dag_output_fetcher.read(timeout)
+            try:
+                result = self._dag_output_fetcher.read(timeout)
+            except RayChannelTimeoutError as e:
+                raise RayChannelTimeoutError(
+                    "If the execution is expected to take a long time, increase "
+                    f"RAY_CGRAPH_get_timeout which is currently {timeout} seconds. "
+                    "Otherwise, this may indicate that the execution is hanging."
+                ) from e
+
             self._max_finished_execution_index += 1
             self._cache_execution_results(
                 self._max_finished_execution_index,