[autoscaler] give max_workers precedence over min_workers in resource demand scheduler (#12106)

python/ray/autoscaler/_private/resource_demand_scheduler.py

@@ -523,7 +523,8 @@ def _add_min_workers_nodes(
     total_nodes_to_add_dict = {}
     for node_type, config in node_types.items():
         existing = node_type_counts.get(node_type, 0)
-        target = config.get("min_workers", 0)
+        target = min(
+            config.get("min_workers", 0), config.get("max_workers", 0))
         if existing < target:
             total_nodes_to_add_dict[node_type] = target - existing
             node_type_counts[node_type] = target

python/ray/tests/test_resource_demand_scheduler.py

@@ -246,6 +246,13 @@ def test_add_min_workers_nodes():
             "min_workers": 99999,
             "max_workers": 99999,
         },
+        "gpubla": {
+            "resources": {
+                "GPU": 1
+            },
+            "min_workers": 10,
+            "max_workers": 0,
+        },
     }
     assert _add_min_workers_nodes([],
                                   {},
@@ -281,6 +288,18 @@ def test_add_min_workers_nodes():
         "gpu": 99999
     }, {})
 
+    assert _add_min_workers_nodes([], {},
+                                  {"gpubla": types["gpubla"]}) == ([], {}, {})
+
+    types["gpubla"]["max_workers"] = 10
+    assert _add_min_workers_nodes([], {}, {"gpubla": types["gpubla"]}) == ([{
+        "GPU": 1
+    }] * 10, {
+        "gpubla": 10
+    }, {
+        "gpubla": 10
+    })
+
 
 def test_get_nodes_to_launch_with_min_workers():
     provider = MockProvider()