code: def test_custom_json_data_source_abort(self):
      data_source = self._get_test_json_data_source()
      self.spark.dataSource.register(data_source)
      with tempfile.TemporaryDirectory() as d:
          with self.assertRaises(PythonException):
              self.spark.range(0, 8, 1, 3).write.format("my-json").mode(
                  "append"
              ).save(d)
          with open(os.path.join(d, "_failed.txt"), "r") as file:
              text = file.read()
          assert text == "failed"
  
imports: import os
  import tempfile
  import unittest
  from typing import Callable, Union
  from pyspark.errors import PythonException
  from pyspark.sql.datasource import DataSource, DataSourceReader, InputPartition, DataSourceWriter, WriterCommitMessage, CaseInsensitiveDict
  from pyspark.sql.types import Row, StructType
  from pyspark.testing import assertDataFrameEqual
  from pyspark.testing.sqlutils import ReusedSQLTestCase
  from pyspark.testing.utils import SPARK_HOME
label_desc: Use image node_id 21 for calling the BasePythonDataSourceTestsMixin obj's underlying member method code with example usage: obj.test_custom_json_data_source_abort() without return types