refactor remove executor from constructor in HashShuffleAggregatorIssueDetector

Signed-off-by: machichima <nary12321@gmail.com>

Author: machichima

python/ray/data/_internal/issue_detection/detectors/__init__.py

@@ -4,6 +4,7 @@
 )
 from ray.data._internal.issue_detection.detectors.hash_shuffle_detector import (
     HashShuffleAggregatorIssueDetector,
+    HashShuffleAggregatorIssueDetectorConfig,
 )
 from ray.data._internal.issue_detection.detectors.high_memory_detector import (
     HighMemoryIssueDetector,
@@ -13,7 +14,8 @@
 __all__ = [
     "HangingExecutionIssueDetector",
     "HangingExecutionIssueDetectorConfig",
+    "HashShuffleAggregatorIssueDetector",
+    "HashShuffleAggregatorIssueDetectorConfig",
     "HighMemoryIssueDetector",
     "HighMemoryIssueDetectorConfig",
-    "HashShuffleAggregatorIssueDetector",
 ]

python/ray/data/_internal/issue_detection/detectors/hash_shuffle_detector.py

@@ -1,5 +1,6 @@
 import time
-from typing import TYPE_CHECKING, List
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Callable, List
 
 import ray
 from ray.data._internal.execution.operators.hash_shuffle import (
@@ -15,15 +16,29 @@
 
 if TYPE_CHECKING:
     from ray.data._internal.execution.streaming_executor import StreamingExecutor
-    from ray.data.context import DataContext
+    from ray.data._internal.execution.interfaces.physical_operator import (
+        PhysicalOperator,
+    )
+
+@dataclass
+class HashShuffleAggregatorIssueDetectorConfig:
+    """Configuration for HashShuffleAggregatorIssueDetector."""
+    detection_time_interval_s: float = 30.0
+    min_wait_time_s: float = 300.0
 
 
 class HashShuffleAggregatorIssueDetector(IssueDetector):
     """Detector for hash shuffle aggregator health issues."""
 
-    def __init__(self, executor: "StreamingExecutor", ctx: "DataContext"):
-        self._executor = executor
-        self._ctx = ctx
+    def __init__(
+        self,
+        dataset_id: str,
+        get_operators_fn: Callable[[], List["PhysicalOperator"]],
+        config: "HashShuffleAggregatorIssueDetectorConfig",
+    ):
+        self._dataset_id = dataset_id
+        self._get_operators = get_operators_fn
+        self._detector_cfg = config
         self._last_warning_times = {}  # Track per-operator warning times
 
     @classmethod
@@ -38,14 +53,24 @@ def from_executor(
         Returns:
             An instance of HashShuffleAggregatorIssueDetector.
         """
-        return cls(executor, executor._data_context)
+        def get_operators_fn() -> List["PhysicalOperator"]:
+            if not executor._topology:
+                return []
+            return list(executor._topology.keys())
+
+        ctx = executor._data_context
+        return cls(
+            dataset_id=executor._dataset_id,
+            get_operators_fn=get_operators_fn,
+            config=ctx.issue_detectors_config.hash_shuffle_detector_config,
+        )
 
     def detect(self) -> List[Issue]:
         issues = []
         current_time = time.time()
 
         # Find all hash shuffle operators in the topology
-        for op in self._executor._topology.keys():
+        for op in self._get_operators():
             if not isinstance(op, HashShuffleOperator):
                 continue
 
@@ -68,7 +93,7 @@ def detect(self) -> List[Issue]:
                 message = self._format_health_warning(aggregator_info)
                 issues.append(
                     Issue(
-                        dataset_name=self._executor._dataset_id,
+                        dataset_name=self._dataset_id,
                         operator_id=op.id,
                         issue_type=IssueType.HANGING,
                         message=message,
@@ -79,7 +104,7 @@ def detect(self) -> List[Issue]:
         return issues
 
     def detection_time_interval_s(self) -> float:
-        return self._ctx.hash_shuffle_aggregator_health_warning_interval_s
+        return self._detector_cfg.detection_time_interval_s
 
     def _should_emit_warning(
         self, op_id: str, current_time: float, info: AggregatorHealthInfo
@@ -93,7 +118,7 @@ def _should_emit_warning(
         # Check if enough time has passed since start
         if (
             current_time - info.started_at
-            < self._ctx.min_hash_shuffle_aggregator_wait_time_in_s
+            < self._detector_cfg.min_wait_time_s
         ):
             return False
 

python/ray/data/_internal/issue_detection/issue_detector_configuration.py

@@ -4,6 +4,8 @@
 from ray.data._internal.issue_detection.detectors import (
     HangingExecutionIssueDetector,
     HangingExecutionIssueDetectorConfig,
+    HashShuffleAggregatorIssueDetector,
+    HashShuffleAggregatorIssueDetectorConfig,
     HighMemoryIssueDetector,
     HighMemoryIssueDetectorConfig,
 )
@@ -15,9 +17,16 @@ class IssueDetectorsConfiguration:
     hanging_detector_config: HangingExecutionIssueDetectorConfig = field(
         default_factory=HangingExecutionIssueDetectorConfig
     )
+    hash_shuffle_detector_config: HashShuffleAggregatorIssueDetectorConfig = field(
+        default_factory=HashShuffleAggregatorIssueDetectorConfig
+    )
     high_memory_detector_config: HighMemoryIssueDetectorConfig = field(
         default_factory=HighMemoryIssueDetectorConfig
     )
     detectors: List[Type[IssueDetector]] = field(
-        default_factory=lambda: [HangingExecutionIssueDetector, HighMemoryIssueDetector]
+        default_factory=lambda: [
+            HangingExecutionIssueDetector,
+            HashShuffleAggregatorIssueDetector,
+            HighMemoryIssueDetector,
+        ]
     )