Repository: ray-project/ray
File: python/ray/train/v2/_internal/execution/controller/controller.py (Python)

Code before:
    ) -> Optional["TrainControllerLoopIterationResult"]:
        optional_error = self._controller_callback_manager.invoke(
            hook_name, *args, **context
        )
        if optional_error:
            failure_decision = self._failure_policy.make_decision(
                training_failed_error=optional_error,
            )
            # Avoid re-entering controller callback hooks while handling a callback failure.
            return self._execute_failure_decision(
                failure_decision,
                training_failed_error=optional_error,
                invoke_failure_decision_callbacks=invoke_failure_decision_callbacks,
            )
        return None

    def _execute_resize_decision(
        self, decision: ResizeDecision
    ) -> TrainControllerLoopIterationResult:
        """Executes resize decisions."""

        optional_result = self._invoke_controller_callback_or_handle_failure(
            "before_controller_execute_resize_decision", decision
        )
        if optional_result:
            return optional_result

        if self._worker_group:
            self._shutdown_worker_group()

        optional_controller_error = self._start_worker_group(
            num_workers=decision.num_workers,
            resources_per_worker=decision.resources_per_worker,
        )

        if optional_controller_error:
            failure_decision = self._failure_policy.make_decision(
                training_failed_error=optional_controller_error,
            )
            return self._execute_failure_decision(
                failure_decision,
                training_failed_error=optional_controller_error,
            )
        else:
            return TrainControllerLoopIterationResult(
                run_attempt_id=self._get_run_attempt_id(),
                previous_state=self._state,
                next_state=RunningState(),
            )

    def _get_retry_state(

Review comment (bug):
what about this instead:

```python
try:
    self._controller_callback_manager.invoke("before_controller_execute_resize_decision", decision)
except ControllerError as e:
    return self._execute_failure_decision(...)
```

Code after:
    ) -> Optional["TrainControllerLoopIterationResult"]:
        """Invoke a named controller hook and catch any exceptions.

        This method invokes all callbacks registered for the given controller hook.
        If a callback raises an error, the error is routed through the failure policy
        and may produce a ``TrainControllerLoopIterationResult``, indicating that the
        current controller step should exit early with this failure result.

        Args:
            hook_name: The controller hook name to invoke.
            *args: Positional arguments to pass to the hook.
            invoke_failure_decision_callbacks: Whether to invoke failure-decision hooks
                when handling a callback failure.
            **context: Keyword arguments to pass to the hook.

        Returns:
            failure_result: A``TrainControllerLoopIterationResult`` if the hook execution results
            in an early exit from the controller loop to raise the callback error,
            or ``None`` if hook execution completes successfully.
        """
        try:
            self._controller_callback_manager.invoke(hook_name, *args, **context)
        except ControllerError as error:
            failure_decision = self._failure_policy.make_decision(
                training_failed_error=error,
            )
            # Avoid re-entering controller callback hooks while handling a callback failure.
            return self._execute_failure_decision(
                failure_decision,
                training_failed_error=error,
                invoke_failure_decision_callbacks=invoke_failure_decision_callbacks,
            )
        return None

    def _execute_resize_decision(
        self, decision: ResizeDecision
    ) -> TrainControllerLoopIterationResult:
        """Executes resize decisions."""

        failure_result = self._run_controller_hook(
            "before_controller_execute_resize_decision", decision
        )
        if failure_result:
            return failure_result

        if self._worker_group:
            self._shutdown_worker_group()

        optional_controller_error = self._start_worker_group(
            num_workers=decision.num_workers,
            resources_per_worker=decision.resources_per_worker,