docstr: Uses port forward to talk to MLMD gRPC server.
docstr_tok: ["Uses","port","forward","to","talk","to","MLMD","gRPC","server","."]
code: def _setup_mlmd_port_forward(cls) -> subprocess.Popen:
      """Uses port forward to talk to MLMD gRPC server."""
      grpc_port = cls._get_grpc_port()
  
      is_bind = False
      forwarded_port = None
  
      for port in range(
          _KFP_E2E_TEST_FORWARDING_PORT_BEGIN,
          _KFP_E2E_TEST_FORWARDING_PORT_END,
      ):
          grpc_forward_command = [
              "kubectl",
              "port-forward",
              "deployment/metadata-grpc-deployment",
              "-n",
              "kubeflow",
              ("%s:%s" % (port, grpc_port)),
          ]
          # Begin port forwarding.
          proc = subprocess.Popen(grpc_forward_command)
          try:
              # Wait while port forward to pod is being established
              poll_grpc_port_command = ["lsof", "-i", ":%s" % port]
              result = subprocess.run(  # pylint: disable=subprocess-run-check
                  poll_grpc_port_command, stdout=subprocess.PIPE
              )
              for _ in range(_MAX_ATTEMPTS):
                  if (
                      result.returncode == 0
                      and "kubectl" in result.stdout.decode("utf-8")
                  ):
                      is_bind = True
                      break
                  logging.info(
                      "Waiting while gRPC port-forward is being established..."
                  )
                  time.sleep(5)
                  result = subprocess.run(  # pylint: disable=subprocess-run-check
                      poll_grpc_port_command, stdout=subprocess.PIPE
                  )
  
          except:  # pylint: disable=bare-except
              # Kill the process in case unexpected error occurred.
              proc.kill()
  
          if is_bind:
              forwarded_port = port
              break
  
      if not is_bind:
          raise RuntimeError(
              "Failed to establish gRPC port-forward to cluster in "
              "the specified range: port %s to %s"
              % (
                  _KFP_E2E_TEST_FORWARDING_PORT_BEGIN,
                  _KFP_E2E_TEST_FORWARDING_PORT_END,
              )
          )
  
      # Establish MLMD gRPC channel.
      forwarding_channel = insecure_channel(
          "localhost:%s" % forwarded_port
      )
      cls._stub = (
          metadata_store_service_pb2_grpc.MetadataStoreServiceStub(
              forwarding_channel
          )
      )
  
      return proc
  
imports: import os
  import subprocess
  import time
  from typing import List
  from absl import logging
  from grpc import insecure_channel
  import tensorflow
  from tfx.dsl.io import fileio
  from tfx.orchestration import test_utils
  from tfx.orchestration.experimental.core.testing import test_dynamic_exec_properties_pipeline
  from tfx.orchestration.kubeflow import test_utils
  from tfx.orchestration.test_pipelines import download_grep_print_pipeline
  from tfx.types import standard_artifacts
  from ml_metadata.proto import metadata_store_pb2
  from ml_metadata.proto import metadata_store_service_pb2
  from ml_metadata.proto import metadata_store_service_pb2_grpc
label_desc: Use image node_id 4 for calling the KubeflowEndToEndTest obj's underlying member method code with example usage: obj._setup_mlmd_port_forward(cls) and returns: proc