code: def check_job_cancellation(
      self,
      setter,
      canceller,
      thread_ids,
      thread_ids_to_cancel,
      thread_ids_to_run,
  ):
      job_id_a = "job_ids_to_cancel"
      job_id_b = "job_ids_to_run"
      threads = []
  
      # 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)
  
              def func(itr):
                  for pdf in itr:
                      time.sleep(pdf._1.iloc[0])
                      yield pdf
  
              self.spark.createDataFrame([[20]]).repartition(
                  1
              ).mapInPandas(func, schema="_1 LONG").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])
      self.spark.clearTags()
  
      # 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(10)
      # 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
              ),
          )
  
imports: import threading
  import time
  import unittest
  from typing import Optional
  from pyspark import InheritableThread, inheritable_thread_target
  from pyspark.sql.connect.client import ChannelBuilder
  from pyspark.sql.connect.session import SparkSession
  from pyspark.testing.connectutils import should_test_connect
label_desc: Use image node_id 4 for calling the JobCancellationTests obj's underlying member method code with example usage: obj.check_job_cancellation(setter, canceller, thread_ids, thread_ids_to_cancel, thread_ids_to_run) without return types