docstr: collection: auto_id, customized_id
  collection schema: [pk, float_vector]
  Steps:
  1. create collection
  2. import data with wrong dimension
  3. verify the data entities is 0 and task was failed
  4. wait for task clean triggered
  5. verify the task was cleaned
docstr_tok: ["collection",":","auto_id",",","customized_id","collection","schema",":","[","pk",",","float_vector","]","Steps",":","1.","create","collection","2.","import","data","with","wrong","dimension","3.","verify","the","data","entities","is","0","and","task","was","failed","4.","wait","for","task","clean","triggered","5.","verify","the","task","was","cleaned"]
code: def test_failed_task_was_cleaned(
      self, is_row_based, auto_id, dim, entities
  ):
      """
      collection: auto_id, customized_id
      collection schema: [pk, float_vector]
      Steps:
      1. create collection
      2. import data with wrong dimension
      3. verify the data entities is 0 and task was failed
      4. wait for task clean triggered
      5. verify the task was cleaned
  
      """
      files = prepare_bulk_insert_json_files(
          minio_endpoint=self.minio_endpoint,
          bucket_name=self.bucket_name,
          is_row_based=is_row_based,
          rows=entities,
          dim=dim,
          auto_id=auto_id,
          data_fields=default_vec_only_fields,
          err_type=DataErrorType.one_entity_wrong_dim,
          wrong_position=entities // 2,
          force=True,
      )
      self._connect()
      c_name = cf.gen_unique_str("bulk_insert")
      fields = [
          cf.gen_int64_field(name=df.pk_field, is_primary=True),
          cf.gen_float_vec_field(name=df.vec_field, dim=dim),
      ]
      schema = cf.gen_collection_schema(fields=fields, auto_id=auto_id)
      self.collection_wrap.init_collection(c_name, schema=schema)
      # import data
      t0 = time.time()
      task_id, _ = self.utility_wrap.do_bulk_insert(
          collection_name=c_name,
          partition_name=None,
          is_row_based=is_row_based,
          files=files,
      )
      logging.info(f"bulk insert task ids:{task_id}")
      (
          success,
          states,
      ) = self.utility_wrap.wait_for_bulk_insert_tasks_completed(
          task_ids=[task_id], timeout=90
      )
      tt = time.time() - t0
      log.info(f"bulk insert state:{success} in {tt}")
      assert not success
      for state in states.values():
          assert state.state_name in ["Failed", "Failed and cleaned"]
  
      num_entities = self.collection_wrap.num_entities
      log.info(f" collection entities: {num_entities}")
      assert num_entities == 0
      log.info("wait for task clean triggered")
      time.sleep(6 * 60)  # wait for 6 minutes for task clean triggered
      num_entities = self.collection_wrap.num_entities
      log.info(f" collection entities: {num_entities}")
      assert num_entities == 0
      (
          success,
          states,
      ) = self.utility_wrap.wait_for_bulk_insert_tasks_completed(
          task_ids=[task_id], timeout=90
      )
      assert not success
      for state in states.values():
          assert state.state_name in ["Failed and cleaned"]
  
imports: import logging
  import time
  import pytest
  from pathlib import Path
  from base.client_base import TestcaseBase
  from common import common_func
  from common import common_type
  from common.milvus_sys import MilvusSys
  from common.common_type import CaseLabel, CheckTasks
  from utils.util_k8s import get_pod_ip_name_pairs, get_milvus_instance_name, get_milvus_deploy_tool
  from utils.util_log import test_log
  from common.bulk_insert_data import prepare_bulk_insert_json_files, DataField, DataErrorType
label_desc: Use image node_id 2 for calling the TestBulkInsertTaskClean obj's underlying member method code with example usage: obj.test_failed_task_was_cleaned(is_row_based, auto_id, dim, entities) without return types