[Data] Deprecate V1 cluster autoscaler in favor of V2 (ray-project#60474)

## Description
This document describes how to verify the V1 cluster autoscaler
deprecation.

1. Added `@Deprecated` annotation to `DefaultClusterAutoscaler` in
`default_cluster_autoscaler.py`
2. Changed default from `V1` to `V2` in `__init__.py`

## Related issues
Closes ray-project#60459

## Additional information
### Testing
**Test V1 via env var emits deprecation warning:**
```bash
RAY_DATA_CLUSTER_AUTOSCALER=V1 python -c "
import warnings
warnings.simplefilter('always')
from unittest.mock import MagicMock
from ray.data._internal.cluster_autoscaler import create_cluster_autoscaler

mock_data_context = MagicMock()
mock_data_context.execution_options.resource_limits = None
autoscaler = create_cluster_autoscaler(MagicMock(), MagicMock(), mock_data_context, execution_id='test')
print(f'Created: {type(autoscaler).__name__}')
"
```

```
...RayDeprecationWarning: This API is deprecated and may be removed in future Ray releases...
DefaultClusterAutoscaler (V1) is deprecated. Use DefaultClusterAutoscalerV2 instead by setting RAY_DATA_CLUSTER_AUTOSCALER=V2 or using the default.
Created: DefaultClusterAutoscaler
```

**Test V2 (default) has no deprecation warning:**
```bash
python -c "
import warnings
warnings.filterwarnings('always', category=DeprecationWarning)
from unittest.mock import MagicMock
from ray.data._internal.cluster_autoscaler import create_cluster_autoscaler

mock_data_context = MagicMock()
mock_data_context.execution_options.resource_limits = None
autoscaler = create_cluster_autoscaler(MagicMock(), MagicMock(), mock_data_context, execution_id='test')
print(f'Created: {type(autoscaler).__name__}')
"
```

```
Created: DefaultClusterAutoscalerV2
```

---------

Signed-off-by: Ryan Huang <ryankert01@gmail.com>
Signed-off-by: 400Ping <jiekaichang@apache.org>

Author: ryankert01

python/ray/data/_internal/cluster_autoscaler/__init__.py

@@ -19,7 +19,7 @@
 
 
 CLUSTER_AUTOSCALER_ENV_KEY = "RAY_DATA_CLUSTER_AUTOSCALER"
-CLUSTER_AUTOSCALER_ENV_DEFAULT_VALUE = "V1"
+CLUSTER_AUTOSCALER_ENV_DEFAULT_VALUE = "V2"
 
 
 class ClusterAutoscalerVersion(Enum):

python/ray/data/_internal/cluster_autoscaler/default_cluster_autoscaler.py

@@ -1,5 +1,6 @@
 import math
 import time
+import warnings
 from typing import TYPE_CHECKING, Dict
 
 import ray
@@ -9,11 +10,17 @@
     get_or_create_autoscaling_requester_actor,
 )
 from ray.data._internal.execution.interfaces import ExecutionResources
+from ray.util.annotations import Deprecated, RayDeprecationWarning
 
 if TYPE_CHECKING:
     from ray.data._internal.execution.streaming_executor_state import Topology
 
 
+@Deprecated(
+    message="DefaultClusterAutoscaler (V1) is deprecated and will be removed "
+    "after June 2026. Use DefaultClusterAutoscalerV2 instead by setting "
+    "RAY_DATA_CLUSTER_AUTOSCALER=V2 or using the default."
+)
 class DefaultClusterAutoscaler(ClusterAutoscaler):
     # Min number of seconds between two autoscaling requests.
     MIN_GAP_BETWEEN_AUTOSCALING_REQUESTS = 20
@@ -25,6 +32,13 @@ def __init__(
         *,
         execution_id: str,
     ):
+        warnings.warn(
+            "DefaultClusterAutoscaler (V1) is deprecated and will be removed "
+            "after June 2026. Use DefaultClusterAutoscalerV2 instead by setting "
+            "RAY_DATA_CLUSTER_AUTOSCALER=V2 or using the default.",
+            RayDeprecationWarning,
+            stacklevel=2,
+        )
         self._topology = topology
         self._resource_limits = resource_limits
         self._execution_id = execution_id

python/ray/train/v2/tests/test_data_resource_cleanup.py

@@ -5,8 +5,8 @@
 import pytest
 
 import ray
-from ray.data._internal.execution.autoscaling_requester import (
-    get_or_create_autoscaling_requester_actor,
+from ray.data._internal.cluster_autoscaler.default_autoscaling_coordinator import (
+    get_or_create_autoscaling_coordinator,
 )
 from ray.data._internal.iterator.stream_split_iterator import (
     SplitCoordinator,
@@ -101,16 +101,15 @@ def test_after_worker_group_shutdown():
 def test_split_coordinator_shutdown_executor(ray_start_4_cpus):
     """Tests that the SplitCoordinator properly requests resources for the data executor and cleans up after it is shutdown"""
 
-    def get_resources_when_updated(requester, prev_requests=None, timeout=3.0):
-        """Retrieve resource requests within the specified timeout. Returns after a new request is made or when the time expires."""
-        prev_requests = prev_requests or {}
+    def get_ongoing_requests(coordinator, timeout=3.0):
+        """Retrieve ongoing requests from the AutoscalingCoordinator."""
         deadline = time.time() + timeout
         requests = {}
         while time.time() < deadline:
             requests = ray.get(
-                requester.__ray_call__.remote(lambda r: r._resource_requests)
+                coordinator.__ray_call__.remote(lambda c: dict(c._ongoing_reqs))
             )
-            if requests != prev_requests:
+            if requests:
                 break
             time.sleep(0.05)
         return requests
@@ -123,34 +122,28 @@ def get_resources_when_updated(requester, prev_requests=None, timeout=3.0):
     )
     ray.get(coord.start_epoch.remote(0))
 
-    # Explicity trigger autoscaling
+    # Explicitly trigger autoscaling
     ray.get(
         coord.__ray_call__.remote(
             lambda coord: coord._executor._cluster_autoscaler.try_trigger_scaling()
         )
     )
 
-    # Collect requests
-    requester = get_or_create_autoscaling_requester_actor()
-    requests = get_resources_when_updated(requester)
+    # Collect requests from the AutoscalingCoordinator
+    coordinator = get_or_create_autoscaling_coordinator()
+    requests = get_ongoing_requests(coordinator)
 
-    # One request made, with non-empty resource bundle
+    # One request made (V2 registers with the coordinator)
     assert len(requests) == 1
-    resource_bundles = list(requests.values())[0][0]
-    assert isinstance(resource_bundles, list)
-    bundle = resource_bundles[0]
-    assert bundle != {}
+    requester_id = list(requests.keys())[0]
+    assert requester_id.startswith("data-")
 
     # Shutdown data executor
     ray.get(coord.shutdown_executor.remote())
 
-    requests = get_resources_when_updated(requester, prev_requests=requests)
-
-    # Old resource request overwritten by new cleanup request
-    assert len(requests) == 1
-    resource_bundles = list(requests.values())[0][0]
-    assert isinstance(resource_bundles, dict)
-    assert resource_bundles == {}
+    # Verify that the request is cancelled (removed from ongoing requests)
+    requests = ray.get(coordinator.__ray_call__.remote(lambda c: dict(c._ongoing_reqs)))
+    assert len(requests) == 0, "Resource request was not cancelled"
 
 
 if __name__ == "__main__":