code: def test_pinned_thread(self):
      threads = []
      exceptions = []
      property_name = "test_property_%s" % PinThreadTests.__name__
      jvm_thread_ids = []
  
      for i in range(10):
  
          def test_local_property():
              jvm_thread_id = (
                  self.sc._jvm.java.lang.Thread.currentThread().getId()
              )
              jvm_thread_ids.append(jvm_thread_id)
  
              # If a property is set in this thread, later it should get the same property
              # within this thread.
              self.sc.setLocalProperty(property_name, str(i))
  
              # 5 threads, 1 second sleep. 5 threads without a sleep.
              time.sleep(i % 2)
  
              try:
                  assert self.sc.getLocalProperty(property_name) == str(
                      i
                  )
  
                  # Each command might create a thread in multi-threading mode in Py4J.
                  # This assert makes sure that the created thread is being reused.
                  assert (
                      jvm_thread_id
                      == self.sc._jvm.java.lang.Thread.currentThread().getId()
                  )
              except Exception as e:
                  exceptions.append(e)
  
          threads.append(threading.Thread(target=test_local_property))
  
      for t in threads:
          t.start()
  
      for t in threads:
          t.join()
  
      for e in exceptions:
          raise e
  
      # Created JVM threads should be 10 because Python thread are 10.
      assert len(set(jvm_thread_ids)) == 10
  
file_path: spark/python/pyspark/tests/test_pin_thread.py
imports: import os
  import time
  import threading
  import unittest
  from pyspark import SparkContext, SparkConf, InheritableThread
label_desc: Use image node_id 3 for calling the PinThreadTests obj's underlying member method code with example usage: obj.test_pinned_thread() without return types