disable StreamingOutputBackpressurePolicy (#42000)

Cherrypick #41999 to 2.9.0. This can fix a regression for small-scale jobs (#41496). But it will also hurt stability for large-scale jobs.

Signed-off-by: Hao Chen <chenh1024@gmail.com>

Author: raulchen

python/ray/data/_internal/execution/backpressure_policy/__init__.py

@@ -11,9 +11,8 @@
 # Default enabled backpressure policies and its config key.
 # Use `DataContext.set_config` to config it.
 # TODO(hchen): Enable ConcurrencyCapBackpressurePolicy by default.
-ENABLED_BACKPRESSURE_POLICIES = [
-    StreamingOutputBackpressurePolicy,
-]
+# TODO(hchen): Enable StreamingOutputBackpressurePolicy by default.
+ENABLED_BACKPRESSURE_POLICIES = []
 ENABLED_BACKPRESSURE_POLICIES_CONFIG_KEY = "backpressure_policies.enabled"
 
 

python/ray/data/tests/test_backpressure_policies.py

@@ -6,6 +6,7 @@
 from unittest.mock import MagicMock, patch
 
 import numpy as np
+import pytest
 
 import ray
 from ray.data._internal.execution.backpressure_policy import (
@@ -409,6 +410,7 @@ def foo(self):
         )
 
 
+@pytest.mark.skip("Re-enable after enabling StreamingOutputBackpressurePolicy.")
 def test_large_e2e_backpressure(shutdown_only, restore_data_context):  # noqa: F811
     """Test backpressure on a synthetic large-scale workload."""
     # The cluster has 10 CPUs and 200MB object store memory.