docstr: collection: auto_id, customized_id
  collection schema: [pk, float_vector]
  Steps:
  1. create collection
  2. import data
  3. verify the data entities equal the import data
  4. load the collection
  5. verify search successfully
  6. verify query successfully
  7. wait for task clean triggered
  8. verify the task not cleaned
docstr_tok: ["collection",":","auto_id",",","customized_id","collection","schema",":","[","pk",",","float_vector","]","Steps",":","1.","create","collection","2.","import","data","3.","verify","the","data","entities","equal","the","import","data","4.","load","the","collection","5.","verify","search","successfully","6.","verify","query","successfully","7.","wait","for","task","clean","triggered","8.","verify","the","task","not","cleaned"]
code: def test_success_task_not_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
      3. verify the data entities equal the import data
      4. load the collection
      5. verify search successfully
      6. verify query successfully
      7. wait for task clean triggered
      8. verify the task not 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,
          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,
          files=files,
      )
      logging.info(f"bulk insert task ids:{task_id}")
      (
          success,
          _,
      ) = 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 success
  
      num_entities = self.collection_wrap.num_entities
      log.info(f" collection entities: {num_entities}")
      assert num_entities == entities
  
      # verify imported data is available for search
      index_params = ct.default_index
      self.collection_wrap.create_index(
          field_name=df.vec_field, index_params=index_params
      )
      self.collection_wrap.load()
      log.info(f"wait for load finished and be ready for search")
      time.sleep(5)
      log.info(
          f"query seg info: {self.utility_wrap.get_query_segment_info(c_name)[0]}"
      )
      nq = 2
      topk = 2
      search_data = cf.gen_vectors(nq, dim)
      search_params = ct.default_search_params
      res, _ = self.collection_wrap.search(
          search_data,
          df.vec_field,
          param=search_params,
          limit=topk,
          check_task=CheckTasks.check_search_results,
          check_items={"nq": nq, "limit": topk},
      )
      for hits in res:
          ids = hits.ids
          results, _ = self.collection_wrap.query(
              expr=f"{df.pk_field} in {ids}"
          )
          assert len(results) == len(ids)
      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 == entities
      res, _ = self.collection_wrap.search(
          search_data,
          df.vec_field,
          param=search_params,
          limit=topk,
          check_task=CheckTasks.check_search_results,
          check_items={"nq": nq, "limit": topk},
      )
      for hits in res:
          ids = hits.ids
          results, _ = self.collection_wrap.query(
              expr=f"{df.pk_field} in {ids}"
          )
          assert len(results) == len(ids)
  
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 1 for calling the TestBulkInsertTaskClean obj's underlying member method code with example usage: obj.test_success_task_not_cleaned(is_row_based, auto_id, dim, entities) without return types