class_docstr: Defines testing harness for pipeline on KubeflowV2DagRunner.
docstr: Trigger the pipeline execution with a specific job ID.
docstr_tok: ["Trigger","the","pipeline","execution","with","a","specific","job","ID","."]
code: def _run_pipeline(
      self,
      pipeline: tfx_pipeline.Pipeline,
      parameter_values: Optional[Dict[str, Any]] = None,
      exit_handler: Optional[base_node.BaseNode] = None,
  ) -> None:
      """Trigger the pipeline execution with a specific job ID."""
      # Ensure cleanup regardless of whether pipeline succeeds or fails.
      self.addCleanup(
          self._delete_pipeline_output,
          pipeline.pipeline_info.pipeline_name,
      )
  
      # Create DAG runner and add exit handler if present.
      v2_dag_runner_config = (
          kubeflow_v2_dag_runner.KubeflowV2DagRunnerConfig(
              default_image=self.container_image
          )
      )
      v2_dag_runner = kubeflow_v2_dag_runner.KubeflowV2DagRunner(
          config=v2_dag_runner_config,
          output_filename=self._output_filename,
      )
      if exit_handler:
          v2_dag_runner.set_exit_handler(exit_handler)
      v2_dag_runner.run(pipeline, write_out=True)
  
      # Create and submit Vertex job.
      self._job_id = pipeline.pipeline_name
      job = pipeline_jobs.PipelineJob(
          display_name=self._job_id,
          job_id=self._job_id,
          template_path=self._output_filename,
          parameter_values=parameter_values,
      )
      job.submit()
  
      # Monitor job status.
      vertex_client_utils.poll_job_status(
          self._job_id,
          _MAX_JOB_EXECUTION_TIME,
          _POLLING_INTERVAL_IN_SECONDS,
      )
  
imports: import datetime
  import os
  import subprocess
  from typing import Any, Dict, List, Optional
  from absl import logging
  from google.cloud import aiplatform
  from google.cloud.aiplatform import pipeline_jobs
  from tfx.dsl.components.base import base_node
  from tfx.orchestration import pipeline
  from tfx.orchestration import test_utils
  from tfx.orchestration.kubeflow.v2 import kubeflow_v2_dag_runner
  from tfx.orchestration.kubeflow.v2 import vertex_client_utils
  from tfx.utils import io_utils
  from tfx.utils import test_case_utils
label_desc: Use image node_id 7 for calling the BaseKubeflowV2Test obj's underlying member method code with example usage: obj._run_pipeline(pipeline, parameter_values, exit_handler) without return types