code: def test_parent_task_id_non_concurrent_actor(shutdown_only):
      ray.init(_system_config=_SYSTEM_CONFIG)
  
      def run_task_in_thread():
          def thd_task():
              @ray.remote
              def thd_task():
                  pass
  
              ray.get(thd_task.remote())
  
          thd = threading.Thread(target=thd_task)
          thd.start()
          thd.join()
  
      @ray.remote
      class Actor:
          def main_task(self):
              run_task_in_thread()
  
      a = Actor.remote()
      ray.get(a.main_task.remote())
  
      def verify():
          tasks = list_tasks()
          expect_parent_task_id = None
          actual_parent_task_id = None
          for task in tasks:
              if "main_task" in task["name"]:
                  expect_parent_task_id = task["task_id"]
              elif "thd_task" in task["name"]:
                  actual_parent_task_id = task["parent_task_id"]
          print(tasks)
          assert actual_parent_task_id is not None
          assert expect_parent_task_id == actual_parent_task_id
  
          return True
  
      wait_for_condition(verify)
  
imports: from collections import defaultdict
  from typing import Dict
  import os
  import pytest
  import sys
  import threading
  import time
  from ray._private.state_api_test_utils import verify_failed_task
  from ray.exceptions import RuntimeEnvSetupError
  from ray.runtime_env import RuntimeEnv
  import ray
  from ray.util.state.common import ListApiOptions, StateResource
  from ray._private.test_utils import raw_metrics, run_string_as_driver, run_string_as_driver_nonblocking, wait_for_condition
  from ray.util.state import StateApiClient, list_tasks
  from ray._private.worker import RayContext
label_desc: Use image node_id 8 for calling a global function with example usage: test_parent_task_id_non_concurrent_actor(shutdown_only) and returns: True