code: def run(self):
      worker = ray._private.worker.global_worker
      worker_id = worker.core_worker.get_actor_id()
  
      while self.current_step <= self.total_steps:
          if not self.stopped:
              print(
                  f"Sleeping {self.interval_s} secs to executing "
                  f"step {self.current_step}"
              )
              time.sleep(self.interval_s)
              self.current_step += 1
              ray_kv._internal_kv_put(
                  f"JOB:{worker_id}", self.current_step, overwrite=True
              )
          else:
              print("Stop called or reached final step.")
              break
  
      self.stopped = True
      ray_kv._internal_kv_put(
          f"JOB:{worker_id}", "DONE", overwrite=True
      )
      return "DONE"
  
file_path: ray/python/ray/experimental/job/example_job/demo_script.py
imports: import argparse
  import time
  import ray
  import ray.experimental.internal_kv
label_desc: Use image node_id 2 for calling the StepActor obj's underlying member method code with example usage: obj.run() and returns: str