code: def test_worker_raylet_resubscription(
      tmp_path, ray_start_regular_with_external_redis
  ):
      # This test is to make sure resubscription in raylet is working.
      # When subscription failed, raylet will not get worker failure error
      # and thus, it won't kill the worker which is fate sharing with the failed
      # one.
  
      @ray.remote
      def long_run():
          from time import sleep
  
          print("LONG_RUN")
          import os
  
          (tmp_path / "long_run.pid").write_text(str(os.getpid()))
          sleep(10000)
  
      @ray.remote
      def bar():
          import os
  
          return (
              os.getpid(),
              # Use runtime env to make sure task is running in a different
              # ray worker
              long_run.options(
                  runtime_env={"env_vars": {"P": ""}}
              ).remote(),
          )
  
      (pid, obj_ref) = ray.get(bar.remote())
  
      long_run_pid = None
  
      def condition():
          nonlocal long_run_pid
          long_run_pid = int((tmp_path / "long_run.pid").read_text())
          return True
  
      wait_for_condition(condition, timeout=5)
  
      # kill the gcs
      ray._private.worker._global_node.kill_gcs_server()
      ray._private.worker._global_node.start_gcs_server()
      # make sure resubscription is done
      # TODO(iycheng): The current way of resubscription potentially will lose
      # worker failure message because we don't ask for the snapshot of worker
      # status for now. We need to fix it.
      sleep(4)
  
      # then kill the owner
      p = psutil.Process(pid)
      p.kill()
  
      # The long_run_pid should exit
      wait_for_pid_to_exit(long_run_pid, 5)
  
imports: import sys
  import os
  import threading
  from time import sleep
  import pytest
  import ray
  from ray._private.utils import get_or_create_event_loop
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
  import ray._private.gcs_utils
  from ray._private import ray_constants
  from ray._private.test_utils import convert_actor_state, enable_external_redis, generate_system_config_map, wait_for_condition, wait_for_pid_to_exit, run_string_as_driver
  from ray.job_submission import JobSubmissionClient, JobStatus
  from ray._raylet import GcsClient
  import psutil
label_desc: Use image node_id 8 for calling a global function with example usage: test_worker_raylet_resubscription(tmp_path, ray_start_regular_with_external_redis) and returns: os,  long_run, True