class_docstr: Integration test for PreemptionCheckpointHandler.
class_docstr_tok: ["Integration","test","for","PreemptionCheckpointHandler","."]
code: def test_passed_in_save_fn(self):
      if _is_oss():
          rpc_layer = "grpc"
      else:
          rpc_layer = "grpc+loas"
  
      checkpoint_dir = os.path.join(self.get_temp_dir(), "fh_ckpt")
      gfile.MakeDirs(checkpoint_dir)
  
      save_fn = lambda: print("Do nothing")
      termination_config = failure_handling.TerminationConfig(
          save_fn=save_fn
      )
  
      has_chief = False
      cluster_spec = multi_worker_test_base.create_cluster_spec(
          has_chief=has_chief, num_workers=CLUSTER_SIZE
      )
      training_started_event = multi_process_runner.manager().Event()
  
      mpr = multi_process_runner.MultiProcessRunner(
          self.worker_fn,
          cluster_spec,
          args=(
              checkpoint_dir,
              cluster_spec,
              "MWMS_multi_worker",
              "checkpoint",
              [training_started_event],
              None,
              None,
              None,
              termination_config,
          ),
          rpc_layer=rpc_layer,
          return_output=True,
          dependence_on_chief=has_chief,
      )
  
      logging.info("Cluster starting.")
      mpr.start()
      while not training_started_event.is_set():
          time.sleep(1)
  
      killed_worker = random.randrange(0, CLUSTER_SIZE)
      logging.info("sending SIGTERM")
      os.kill(
          mpr.get_process_id("worker", killed_worker), signal.SIGTERM
      )
      logging.info("SIGTERM sent")
  
      # 5 is the grace period length
      raise_if_not_all_exit(5, mpr)
  
      match_group = [
          re.search(r".*ckpt-(\d+).index", a_file)
          for a_file in gfile.ListDirectory(checkpoint_dir)
      ]
  
      # By default, as tested by other test cases, checkpoint will be saved.
      # This passed in save_fn skips it.
      self.assertEmpty(match_group)
  
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 6 for calling the PreemptionCheckpointTest obj's underlying member method code with example usage: obj.test_passed_in_save_fn() without return types