class_docstr: Integration test for PreemptionCheckpointHandler.
class_docstr_tok: ["Integration","test","for","PreemptionCheckpointHandler","."]
code: def test_error_propagation(self):
      error_worker = random.randint(0, CLUSTER_SIZE)
      cluster_spec = multi_worker_test_base.create_cluster_spec(
          has_chief=False, num_workers=CLUSTER_SIZE
      )
      checkpoint_dir = self.get_temp_dir()
  
      def assert_raise_error():
          # Asserts that an error raised during a training step on one of the worker
          # is caught on all workers.
          with self.assertRaises(
              errors_impl.ResourceExhaustedError
          ) as error:
              self.worker_fn(
                  checkpoint_dir,
                  cluster_spec,
                  "MWMS_multi_worker",
                  raise_app_error_on_worker=error_worker,
              )
          self.assertIn(
              "Running out of resources", str(error.exception)
          )
  
      if _is_oss():
          rpc_layer = "grpc"
      else:
          rpc_layer = "grpc+loas"
  
      mpr = multi_process_runner.MultiProcessRunner(
          assert_raise_error,
          cluster_spec,
          rpc_layer=rpc_layer,
          return_output=True,
          dependence_on_chief=False,
      )
  
      logging.info("Cluster starting.")
      mpr.start()
      mpr.join(timeout=250)
  
imports: import os
  import random
  import re
  import signal
  import sys
  import threading
  import time
  from absl.testing import parameterized
  from tensorflow.python.checkpoint import checkpoint
  from tensorflow.python.checkpoint import checkpoint_management
  from tensorflow.python.distribute import collective_all_reduce_strategy
  from tensorflow.python.distribute import combinations
  from tensorflow.python.distribute import distribute_lib
  from tensorflow.python.distribute import mirrored_strategy
  from tensorflow.python.distribute import multi_process_runner
  from tensorflow.python.distribute import multi_worker_test_base
  from tensorflow.python.distribute import multi_worker_util
  from tensorflow.python.distribute import one_device_strategy
  from tensorflow.python.distribute import test_util
  from tensorflow.python.distribute.failure_handling import failure_handling
  from tensorflow.python.distribute.failure_handling import failure_handling_util
  from tensorflow.python.eager import def_function
  from tensorflow.python.framework import config
  from tensorflow.python.framework import constant_op
  from tensorflow.python.framework import errors_impl
  from tensorflow.python.module import module
  from tensorflow.python.ops import variables
  from tensorflow.python.platform import gfile
  from tensorflow.python.platform import test
  from tensorflow.python.platform import tf_logging
  from tensorflow.python.training import server_lib
label_desc: Use image node_id 4 for calling the PreemptionCheckpointTest obj's underlying member method code with example usage: obj.test_error_propagation() without return types