code: def testAsyncCheckpointHookEnabled(self):
      resolver = tpu_cluster_resolver.TPUClusterResolver(
          tpu=FLAGS.tpu, zone=FLAGS.zone, project=FLAGS.project
      )
  
      checkpoint_interval = 5
      config = tpu_config.RunConfig(
          master=resolver.master(),
          model_dir=os.path.join(FLAGS.model_dir, "runconfig"),
          save_checkpoints_steps=1000,
          keep_checkpoint_max=11,  # off by one
          tpu_config=tpu_config.TPUConfig(
              iterations_per_loop=checkpoint_interval,
          ),
      )
  
      estimator = tpu_estimator.TPUEstimator(
          use_tpu=True,
          model_fn=model_fn,
          config=config,
          train_batch_size=32,
          eval_batch_size=32,
          predict_batch_size=1,
          params={},
      )
  
      max_steps = 100
      mock_listener = test.mock.create_autospec(
          basic_session_run_hooks.CheckpointSaverListener
      )
      estimator.train(
          input_fn=input_fn,
          max_steps=max_steps,
          hooks=[
              async_checkpoint.AsyncCheckpointSaverHook(
                  FLAGS.model_dir,
                  save_steps=checkpoint_interval,
                  listeners=[mock_listener],
              )
          ],
      )
  
      current_step = (
          estimator_lib._load_global_step_from_checkpoint_dir(
              FLAGS.model_dir
          )
      )  # pylint: disable=protected-access
  
      # TODO(power) -- identify a better way to count the number of checkpoints.
      checkpoints = file_io.get_matching_files(
          FLAGS.model_dir + "/model.ckpt*.meta"
      )
      checkpoint_count = len(checkpoints)
      logging.info(
          "Found %d checkpoints: %s", checkpoint_count, checkpoints
      )
      self.assertLessEqual(checkpoint_count, 10)
      self.assertEqual(current_step, max_steps)
      mock_listener.before_save.assert_called()
      mock_listener.after_save.assert_called()
  
      # save called by hook in `after_create_session` and every `after_run`
      num_save_calls = 1 + max_steps // checkpoint_interval
      sync_count_1, async_count_1 = _get_checkpoint_metrics_counts()
      # save might be called one extra time in `end` hook based on timing of
      # `_last_checkpoint_step` update in the final `after_run` call
      self.assertIn(sync_count_1, [num_save_calls, num_save_calls + 1])
      self.assertLessEqual(async_count_1, num_save_calls)
      training_time_saved = metrics.GetTrainingTimeSaved(
          api_label=async_checkpoint._ASYNC_CHECKPOINT_V1
      )
      self.assertGreater(training_time_saved, 0)
  
imports: import os
  import numpy
  from tensorflow.core.framework import summary_pb2
  from tensorflow.python.compat import v2_compat
  from tensorflow.python.distribute.cluster_resolver import tpu_cluster_resolver
  from tensorflow.python.framework import constant_op
  from tensorflow.python.framework import dtypes
  from tensorflow.python.lib.io import file_io
  from tensorflow.python.ops import math_ops
  from tensorflow.python.ops import metrics
  from tensorflow.python.ops import variable_scope
  from tensorflow.python.ops.losses import losses
  from tensorflow.python.platform import flags
  from tensorflow.python.platform import test
  from tensorflow.python.platform import tf_logging
  from tensorflow.python.saved_model.pywrap_saved_model import metrics
  from tensorflow.python.tpu import async_checkpoint
  from tensorflow.python.tpu import tpu_optimizer
  from tensorflow.python.training import basic_session_run_hooks
  from tensorflow.python.training import training
  from tensorflow_estimator.python.estimator import estimator
  from tensorflow_estimator.python.estimator import model_fn
  from tensorflow_estimator.python.estimator.tpu import tpu_config
  from tensorflow_estimator.python.estimator.tpu import tpu_estimator
label_desc: Use image node_id 1 for calling the AsyncCheckpointingTest obj's underlying member method code with example usage: obj.testAsyncCheckpointHookEnabled() without return types