code: def test_load_async(
      printer: Callable,
      client: cx.Client,
      api: str,
      load_config: Dict[str, Union[int, float]],
      deploy_timeout: int = None,
      poll_sleep_seconds: int = 1,
      api_config_name: str = "cortex_cpu.yaml",
      node_groups: List[str] = [],
  ):
      total_requests = load_config["total_requests"]
      desired_replicas = load_config["desired_replicas"]
      concurrency = load_config["concurrency"]
      submit_timeout = load_config["submit_timeout"]
      workload_timeout = load_config["workload_timeout"]
  
      api_dir = TEST_APIS_DIR / api
      with open(str(api_dir / api_config_name)) as f:
          api_specs = yaml.safe_load(f)
      assert len(api_specs) == 1
  
      api_specs[0]["autoscaling"] = {
          "min_replicas": desired_replicas,
          "max_replicas": desired_replicas,
      }
      if len(node_groups) > 0:
          api_specs[0]["node_groups"] = node_groups
  
      api_name = api_specs[0]["name"]
      client.deploy(api_spec=api_specs[0])
  
      request_stopper = td.Event()
      map_stopper = td.Event()
      responses: List[Dict[str, Any]] = []
      failed = False
      try:
          printer(f"getting {desired_replicas} replicas ready")
          assert apis_ready(
              client=client,
              api_names=[api_name],
              timeout=deploy_timeout,
          ), f"api {api_name} not ready"
  
          with open(str(api_dir / "sample.json")) as f:
              payload = json.load(f)
  
          printer("start making prediction requests concurrently")
          threads_futures = make_requests_concurrently(
              client,
              api_name,
              concurrency,
              request_stopper,
              responses=responses,
              max_total_requests=total_requests,
              payload=payload,
          )
          assert wait_on_event(
              request_stopper, submit_timeout
          ), f"{total_requests} couldn't be submitted in {submit_timeout}s"
          check_futures_healthy(threads_futures)
          wait_on_futures(threads_futures)
  
          printer("finished making prediction requests")
          assert (
              len(responses) == total_requests
          ), f"the submitted number of requests doesn't match the returned number of responses"
  
          job_ids = []
          for response in responses:
              response_json = response.json()
              assert "id" in response_json
              job_ids.append(response_json["id"])
  
          # assert the results
          printer("start retrieving the async results concurrently")
          results = []
          retrieve_results_concurrently(
              client,
              api_name,
              concurrency,
              map_stopper,
              job_ids,
              results,
              poll_sleep_seconds,
              workload_timeout,
          )
          for request_id, json_result in results:
              # validate keys are in the result json response
              assert (
                  "id" in json_result
              ), f"id key was not present in result response (response: {json_result})"
              assert (
                  "result" in json_result
              ), f"result key was not present in result response (response: {json_result})"
              assert (
                  "timestamp" in json_result
              ), f"timestamp key was not present in result response (response: {json_result})"
  
              # validate result json response has valid values
              assert (
                  json_result["id"] == request_id
              ), f"result 'id' and request 'id' mismatch ({json_result['id']} != {request_id})"
              assert (
                  json_result["status"] == "completed"
              ), f"async workload did not complete (response: {json_result})"
              assert (
                  json_result["timestamp"] != ""
              ), "result 'timestamp' value was empty"
              assert (
                  json_result["result"] != ""
              ), "result 'result' value was empty"
  
      except:
          # best effort
          failed = True
          try:
              api_info = client.get_api(api_name)
              printer(json.dumps(api_info, indent=2))
          finally:
              raise
  
      finally:
          if "results" in vars() and len(results) < total_requests:
              printer(
                  f"{len(results)}/{total_requests} have been successfully retrieved"
              )
          map_stopper.set()
          delete_apis(client, [api_name])
          if failed:
              time.sleep(30)
  
imports: import json
  import math
  import os
  import re
  import threading
  import time
  from http import HTTPStatus
  from pathlib import Path
  from typing import Callable, Dict, Any, List, Union
  import boto3
  import cortex
  import requests
  import yaml
  from e2e.expectations import parse_expectations, assert_response_expectations, assert_json_expectations
  from e2e.generator import load_generator
  from e2e.utils import apis_ready, api_updated, wait_on_event, wait_on_futures, endpoint_ready, post_request, get_request, job_done, jobs_done, request_batch_prediction, request_task, retrieve_async_result, make_requests_concurrently, check_futures_healthy, retrieve_results_concurrently, stream_api_logs, stream_job_logs, wait_for
label_desc: Use image node_id 8 for calling a global function with example usage: test_load_async(printer, client, api, load_config, deploy_timeout, poll_sleep_seconds, api_config_name, node_groups) without return types