code: def test_autologging_slow_api_requests(
      spark_session, format_to_file_path
  ):
      import mlflow.utils.rest_utils
  
      orig = mlflow.utils.rest_utils.http_request
  
      def _slow_api_req_mock(*args, **kwargs):
          if kwargs.get("method") == "POST":
              time.sleep(1)
          return orig(*args, **kwargs)
  
      mlflow.spark.autolog()
      with mlflow.start_run():
          # Mock slow API requests to log Spark datasource information
          with mock.patch(
              "mlflow.utils.rest_utils.http_request"
          ) as http_request_mock:
              http_request_mock.side_effect = _slow_api_req_mock
              run_id = mlflow.active_run().info.run_id
              for data_format, file_path in format_to_file_path.items():
                  df = (
                      spark_session.read.format(data_format)
                      .option("header", "true")
                      .option("inferSchema", "true")
                      .load(file_path)
                  )
                  df.collect()
          # Sleep a bit prior to ending the run to guarantee that the Python process can pick up on
          # datasource read events (simulate the common case of doing work, e.g. model training,
          # on the DataFrame after reading from it)
          time.sleep(1)
  
      # Python subscriber threads should pick up the active run at the time they're notified
      # & make API requests against that run, even if those requests are slow.
      time.sleep(5)
      run = mlflow.get_run(run_id)
      assert _SPARK_TABLE_INFO_TAG_NAME in run.data.tags
      table_info_tag = run.data.tags[_SPARK_TABLE_INFO_TAG_NAME]
      assert table_info_tag == "\n".join(
          [
              _get_expected_table_info_row(path, data_format)
              for data_format, path in format_to_file_path.items()
          ]
      )
  
file_path: mlflow/tests/spark/autologging/datasource/test_spark_datasource_autologging.py
imports: import time
  from unittest import mock
  from pyspark.sql import Row
  from pyspark.sql.types import IntegerType, StructField, StructType
  import mlflow
  import mlflow.spark
  from mlflow._spark_autologging import _SPARK_TABLE_INFO_TAG_NAME
  from mlflow.utils.validation import MAX_TAG_VAL_LENGTH
  from tests.spark.autologging.utils import _assert_spark_data_logged
  from tests.tracking.integration_test_utils import _init_server
label_desc: Use image node_id 8 for calling a global function with example usage: test_autologging_slow_api_requests(spark_session, format_to_file_path) and returns: orig