code: def test_apply_in_pandas_with_state_python_worker_random_failure(
      self,
  ):
      input_path = tempfile.mkdtemp()
      output_path = tempfile.mkdtemp()
      checkpoint_loc = tempfile.mkdtemp()
  
      shutil.rmtree(output_path)
      shutil.rmtree(checkpoint_loc)
  
      def prepare_test_resource():
          data_range = list(string.ascii_lowercase)
          for i in range(5):
              picked_data = [
                  data_range[random.randrange(0, len(data_range) - 1)]
                  for x in range(100)
              ]
  
              with open(input_path + "/part-%i.txt" % i, "w") as fw:
                  for data in picked_data:
                      fw.write(data + "\n")
  
      def run_query():
          df = (
              self.spark.readStream.format("text")
              .option("maxFilesPerTrigger", "1")
              .load(input_path)
          )
  
          for q in self.spark.streams.active:
              q.stop()
          self.assertTrue(df.isStreaming)
  
          output_type = StructType(
              [
                  StructField("value", StringType()),
                  StructField("count", LongType()),
              ]
          )
          state_type = StructType([StructField("cnt", LongType())])
  
          def func(key, pdf_iter, state):
              assert isinstance(state, GroupState)
  
              # user function call will happen at most 26 times
              # should be huge enough to not trigger kill in every batches
              # but should be also reasonable to trigger kill multiple times across batches
              if random.randrange(30) == 1:
                  sys.exit(1)
  
              count = state.getOption
              if count is None:
                  count = 0
              else:
                  count = count[0]
  
              for pdf in pdf_iter:
                  count += len(pdf)
  
              state.update((count,))
              yield pd.DataFrame({"value": [key[0]], "count": [count]})
  
          query = (
              df.groupBy(df["value"])
              .applyInPandasWithState(
                  func,
                  output_type,
                  state_type,
                  "Append",
                  GroupStateTimeout.NoTimeout,
              )
              .writeStream.queryName("this_query")
              .format("json")
              .outputMode("append")
              .option("path", output_path)
              .option("checkpointLocation", checkpoint_loc)
              .start()
          )
  
          return query
  
      prepare_test_resource()
  
      expected = (
          self.spark.read.format("text")
          .load(input_path)
          .groupBy("value")
          .count()
          .sort("value")
          .collect()
      )
  
      q = run_query()
      self.assertEqual(q.name, "this_query")
      self.assertTrue(q.isActive)
  
      def assert_test():
          nonlocal q
          if not q.isActive:
              print("query has been terminated, rerunning query...")
  
              # rerunning query as the query may have been killed by killed python worker
              q = run_query()
  
              self.assertEqual(q.name, "this_query")
              self.assertTrue(q.isActive)
  
          curr_status = q.status
          if (
              not curr_status["isDataAvailable"]
              and not curr_status["isTriggerActive"]
          ):
              # The query is active but not running due to no further data available
              # Check the output now.
              result = (
                  self.spark.read.schema("value string, count int")
                  .format("json")
                  .load(output_path)
                  .groupBy("value")
                  .max("count")
                  .selectExpr("value", "`max(count)` AS count")
                  .sort("value")
                  .collect()
              )
  
              return result == expected
          else:
              # still processing the data, defer checking the output.
              return False
  
      try:
          eventually(timeout=120)(assert_test)()
      finally:
          q.stop()
  
imports: import random
  import shutil
  import string
  import sys
  import tempfile
  import unittest
  from typing import cast
  from pyspark import SparkConf
  from pyspark.sql.streaming.state import GroupStateTimeout, GroupState
  from pyspark.sql.types import LongType, StringType, StructType, StructField, Row
  from pyspark.testing.sqlutils import ReusedSQLTestCase, have_pandas, have_pyarrow, pandas_requirement_message, pyarrow_requirement_message
  from pyspark.testing.utils import eventually
label_desc: Use image node_id 9 for calling the GroupedApplyInPandasWithStateTestsMixin obj's underlying member method code with example usage: obj.test_apply_in_pandas_with_state_python_worker_random_failure() and returns: query, unknown, False