code: def test_failure_recovery(
      ray_start_4_cpus_2_gpus_extra,
      resource_manager_cls,
      max_failures_persistent,
  ):
      """Test failure recover with `max_failures`.
  
      Trials should be retried up to `max_failures` times.
  
      Legacy test: test_trial_runner_2.py::TrialRunnerTest::testFailureRecoveryDisabled
      Legacy test: test_trial_runner_2.py::TrialRunnerTest::testFailureRecoveryEnabled
      Legacy test: test_trial_runner_2.py::TrialRunnerTest::testFailureRecoveryMaxFailures
      """
      max_failures, persistent_error = max_failures_persistent
      searchalg, scheduler = create_mock_components()
  
      runner = TuneController(
          search_alg=searchalg,
          scheduler=scheduler,
          resource_manager_factory=lambda: resource_manager_cls(),
          storage=STORAGE,
      )
      kwargs = {
          "placement_group_factory": PlacementGroupFactory(
              [{"CPU": 1, "GPU": 1}]
          ),
          "stopping_criterion": {"training_iteration": 2},
          "checkpoint_config": CheckpointConfig(checkpoint_frequency=1),
          "max_failures": max_failures,
          "config": {
              "mock_error": True,
              "persistent_error": persistent_error,
          },
          "storage": STORAGE,
      }
      runner.add_trial(Trial("__fake", **kwargs))
      trials = runner.get_trials()
  
      while not runner.is_finished():
          runner.step()
  
      if persistent_error or not max_failures:
          assert trials[0].status == Trial.ERROR
  
          num_failures = max_failures + 1
          assert trials[0].num_failures == num_failures
          # search alg receives on_complete, so only after the max failures
          # have been exhausted. Thus, it only has errored_trials if the
          # trial fails even in the last try.
          assert len(searchalg.errored_trials) == 1
          # search alg receives on_error, so every failure is registered.
          assert len(scheduler.errored_trials) == num_failures
      else:
          assert trials[0].status == Trial.TERMINATED
          assert trials[0].num_failures == 1
          assert len(searchalg.errored_trials) == 0
          assert len(scheduler.errored_trials) == 1
  
imports: import os
  from collections import Counter
  import pytest
  import sys
  import ray
  from ray.train import CheckpointConfig
  from ray.air.execution import FixedResourceManager, PlacementGroupResourceManager
  from ray.tune import PlacementGroupFactory, TuneError
  from ray.tune.execution.tune_controller import TuneController
  from ray.tune.experiment import Trial
  from ray.tune.registry import TRAINABLE_CLASS, _global_registry
  from ray.tune.schedulers import FIFOScheduler
  from ray.tune.search import BasicVariantGenerator
  from ray.train.tests.util import mock_storage_context
  from ray.tune.tests.execution.utils import BudgetResourceManager
label_desc: Use image node_id 5 for calling a global function with example usage: test_failure_recovery(ray_start_4_cpus_2_gpus_extra, resource_manager_cls, max_failures_persistent) without return types