[Tune] Fix ResourceChangingScheduler dropping PGF args

python/ray/air/tests/test_resource_changing.py

@@ -55,6 +55,7 @@ def training_loop(self) -> None:
         scaling_config = self._validate_scaling_config(self.scaling_config)
         pgf = scaling_config.as_placement_group_factory()
         tr = session.get_trial_resources()
+        assert pgf.strategy == "SPREAD"
         assert pgf == tr, (pgf, tr)
         return super().training_loop()
 
@@ -65,14 +66,18 @@ def _ray_params(self):
         scaling_config = self._validate_scaling_config(self.scaling_config)
         pgf = scaling_config.as_placement_group_factory()
         tr = session.get_trial_resources()
+        assert pgf.strategy == "SPREAD"
         assert pgf == tr, (scaling_config, pgf, tr)
         return super()._ray_params
 
 
 def test_data_parallel_trainer(ray_start_8_cpus):
     num_workers = 2
     trainer = AssertingDataParallelTrainer(
-        train_fn, scaling_config=ScalingConfig(num_workers=num_workers)
+        train_fn,
+        scaling_config=ScalingConfig(
+            num_workers=num_workers, placement_strategy="SPREAD"
+        ),
     )
     tuner = Tuner(
         trainer,
@@ -108,7 +113,7 @@ def test_gbdt_trainer(ray_start_8_cpus):
     trainer = AssertingXGBoostTrainer(
         datasets={TRAIN_DATASET_KEY: train_ds},
         label_column="target",
-        scaling_config=ScalingConfig(num_workers=2),
+        scaling_config=ScalingConfig(num_workers=2, placement_strategy="SPREAD"),
         params={
             "objective": "binary:logistic",
             "eval_metric": ["logloss"],

python/ray/train/gbdt_trainer.py

@@ -12,6 +12,7 @@
 from ray.tune.trainable.util import TrainableUtil
 from ray.util.annotations import DeveloperAPI
 from ray._private.dict import flatten_dict
+from ray.tune.execution.placement_groups import PlacementGroupFactory
 
 if TYPE_CHECKING:
     import xgboost_ray
@@ -298,8 +299,25 @@ def default_resource_request(cls, config):
                 validated_scaling_config = trainer_cls._validate_scaling_config(
                     updated_scaling_config
                 )
-                return _convert_scaling_config_to_ray_params(
+                # RayParams PGF drops strategy, args and kwargs so we have to get them
+                # from ScalingConfig PGF and then create a new PGF combining it
+                # all together.
+                pgf_from_scaling_config = (
+                    validated_scaling_config.as_placement_group_factory()
+                )
+                pgf_from_ray_params = _convert_scaling_config_to_ray_params(
                     validated_scaling_config, ray_params_cls, default_ray_params
                 ).get_tune_resources()
+                new_bundles = pgf_from_ray_params.bundles
+                final_pgf = PlacementGroupFactory(
+                    new_bundles,
+                    strategy=pgf_from_scaling_config.strategy,
+                    *pgf_from_scaling_config._args,
+                    **pgf_from_scaling_config._kwargs,
+                )
+                final_pgf._head_bundle_is_empty = (
+                    pgf_from_ray_params._head_bundle_is_empty
+                )
+                return final_pgf
 
         return GBDTTrainable

python/ray/tune/schedulers/resource_changing_scheduler.py

@@ -451,7 +451,12 @@ def __call__(
             base_bundles, added_bundles, increase_by, False
         )
 
-        pgf = PlacementGroupFactory(new_bundles)
+        pgf = PlacementGroupFactory(
+            new_bundles,
+            strategy=base_trial_resource.strategy,
+            *base_trial_resource._args,
+            **base_trial_resource._kwargs,
+        )
         pgf._head_bundle_is_empty = base_trial_resource._head_bundle_is_empty
         return pgf