code: def check_job_cancellation(self, setter, canceller):
      job_id_a = "job_ids_to_cancel"
      job_id_b = "job_ids_to_run"
  
      threads = []
      thread_ids = range(4)
      thread_ids_to_cancel = [i for i in thread_ids if i % 2 == 0]
      thread_ids_to_run = [i for i in thread_ids if i % 2 != 0]
  
      # A list which records whether job is cancelled.
      # The index of the array is the thread index which job run in.
      is_job_cancelled = [False for _ in thread_ids]
  
      def run_job(job_id, index):
          """
          Executes a job with the group ``job_group``. Each job waits for 3 seconds
          and then exits.
          """
          try:
              setter(job_id)
              self.sc.parallelize([15]).map(
                  lambda x: time.sleep(x)
              ).collect()
              is_job_cancelled[index] = False
          except Exception:
              # Assume that exception means job cancellation.
              is_job_cancelled[index] = True
  
      # Test if job succeeded when not cancelled.
      run_job(job_id_a, 0)
      self.assertFalse(is_job_cancelled[0])
  
      # Run jobs
      for i in thread_ids_to_cancel:
          t = threading.Thread(target=run_job, args=(job_id_a, i))
          t.start()
          threads.append(t)
  
      for i in thread_ids_to_run:
          t = threading.Thread(target=run_job, args=(job_id_b, i))
          t.start()
          threads.append(t)
  
      # Wait to make sure all jobs are executed.
      time.sleep(3)
      # And then, cancel one job group.
      canceller(job_id_a)
  
      # Wait until all threads launching jobs are finished.
      for t in threads:
          t.join()
  
      for i in thread_ids_to_cancel:
          self.assertTrue(
              is_job_cancelled[i],
              "Thread {i}: Job in group A was not cancelled.".format(
                  i=i
              ),
          )
  
      for i in thread_ids_to_run:
          self.assertFalse(
              is_job_cancelled[i],
              "Thread {i}: Job in group B did not succeeded.".format(
                  i=i
              ),
          )
  
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 6 for calling the PinThreadTests obj's underlying member method code with example usage: obj.check_job_cancellation(setter, canceller) without return types