code: def inheritable_thread_target(
      f: Optional[Union[Callable, "SparkSession"]] = None
  ) -> Callable:
      """
      Return thread target wrapper which is recommended to be used in PySpark when the
      pinned thread mode is enabled. The wrapper function, before calling original
      thread target, it inherits the inheritable properties specific
      to JVM thread such as ``InheritableThreadLocal``, or thread local such as tags
      with Spark Connect.
  
      When the pinned thread mode is off, it return the original ``f``.
  
      .. versionadded:: 3.2.0
  
      .. versionchanged:: 3.5.0
          Supports Spark Connect.
  
      Parameters
      ----------
      f : function, or :class:`SparkSession`
          the original thread target, or :class:`SparkSession` if Spark Connect is being used.
          See the examples below.
  
      Notes
      -----
      This API is experimental.
  
      It is important to know that it captures the local properties or tags when you
      decorate it whereas :class:`InheritableThread` captures when the thread is started.
      Therefore, it is encouraged to decorate it when you want to capture the local
      properties.
  
      For example, the local properties or tags from the current Spark context or Spark
      session is captured when you define a function here instead of the invocation:
  
      >>> @inheritable_thread_target
      ... def target_func():
      ...     pass  # your codes.
  
      If you have any updates on local properties or tags afterwards, it would not be
      reflected to the Spark context in ``target_func()``.
  
      The example below mimics the behavior of JVM threads as close as possible:
  
      >>> Thread(target=inheritable_thread_target(target_func)).start()  # doctest: +SKIP
  
      If you're using Spark Connect, you should explicitly provide Spark session as follows:
  
      >>> @inheritable_thread_target(session)  # doctest: +SKIP
      ... def target_func():
      ...     pass  # your codes.
  
      >>> Thread(target=inheritable_thread_target(session)(target_func)).start()  # doctest: +SKIP
      """
      from pyspark.sql import is_remote
  
      # Spark Connect
      if is_remote():
          session = f
          assert (
              session is not None
          ), "Spark Connect session must be provided."
  
          def outer(ff: Callable) -> Callable:
              session_client_thread_local_attrs = [
                  (attr, copy.deepcopy(value))
                  for (
                      attr,
                      value,
                  ) in session.client.thread_local.__dict__.items()  # type: ignore[union-attr]
              ]
  
              @functools.wraps(ff)
              def inner(*args: Any, **kwargs: Any) -> Any:
                  # Set thread locals in child thread.
                  for attr, value in session_client_thread_local_attrs:
                      setattr(session.client.thread_local, attr, value)  # type: ignore[union-attr]
                  return ff(*args, **kwargs)
  
              return inner
  
          return outer
  
      # Non Spark Connect
      from pyspark import SparkContext
  
      if isinstance(SparkContext._gateway, ClientServer):
          # Here's when the pinned-thread mode (PYSPARK_PIN_THREAD) is on.
  
          # NOTICE the internal difference vs `InheritableThread`. `InheritableThread`
          # copies local properties when the thread starts but `inheritable_thread_target`
          # copies when the function is wrapped.
          assert SparkContext._active_spark_context is not None
          properties = (
              SparkContext._active_spark_context._jsc.sc()
              .getLocalProperties()
              .clone()
          )
          assert callable(f)
  
          @functools.wraps(f)
          def wrapped(*args: Any, **kwargs: Any) -> Any:
              # Set local properties in child thread.
              assert SparkContext._active_spark_context is not None
              SparkContext._active_spark_context._jsc.sc().setLocalProperties(
                  properties
              )
              return f(*args, **kwargs)  # type: ignore[misc, operator]
  
          return wrapped
      else:
          return f
  
imports: import copy
  import functools
  import itertools
  import os
  import platform
  import re
  import sys
  import threading
  import traceback
  import typing
  from types import TracebackType
  from typing import Any, Callable, IO, Iterator, List, Optional, TextIO, Tuple, Union
  from py4j.clientserver import ClientServer
  from pyspark.errors import PySparkRuntimeError
  from py4j.java_gateway import JavaObject
label_desc: Use image node_id 7 for calling a global function with example usage: inheritable_thread_target(f) and returns: outer, wrapped, f, inner, f, ff