class_docstr: Responsible for launching a container executor.
class_docstr_tok: ["Responsible","for","launching","a","container","executor","."]
docstr: Execute underlying component implementation.
docstr_tok: ["Execute","underlying","component","implementation","."]
code: def run_executor(
      self, execution_info: data_types.ExecutionInfo
  ) -> execution_result_pb2.ExecutorOutput:
      """Execute underlying component implementation."""
  
      context = placeholder_utils.ResolutionContext(
          exec_info=execution_info,
          executor_spec=self._executor_spec,
          platform_config=self._platform_config,
          pipeline_platform_config=self._pipeline_platform_config,
      )
  
      component_executor_spec = (
          executor_specs.TemplatedExecutorContainerSpec(
              image=self._container_executor_spec.image,
              command=[
                  placeholder_utils.resolve_placeholder_expression(
                      cmd, context
                  )
                  for cmd in self._container_executor_spec.commands
              ],
          )
      )
  
      logging.info("Container spec: %s", vars(component_executor_spec))
      logging.info(
          "Docker platform config: %s",
          proto_utils.proto_to_json(self._docker_platform_config),
      )
  
      # Call client.containers.run and wait for completion.
      # ExecutorContainerSpec follows k8s container spec which has different
      # names to Docker's container spec. It's intended to set command to docker's
      # entrypoint and args to docker's command.
      if self._docker_platform_config.docker_server_url:
          client = docker.DockerClient(
              base_url=self._docker_platform_config.docker_server_url
          )
      else:
          client = docker.from_env()
  
      run_args = self._build_run_args(self._docker_platform_config)
      container = client.containers.run(
          image=component_executor_spec.image,
          command=component_executor_spec.command,
          detach=True,
          **run_args,
      )
  
      # Streaming logs
      for log in container.logs(stream=True):
          logging.info("Docker: %s", log.decode("utf-8"))
      exit_code = container.wait()["StatusCode"]
      if exit_code != 0:
          raise RuntimeError(
              'Container exited with error code "{}"'.format(exit_code)
          )
      # TODO(b/141192583): Report data to publisher
      # - report container digest
      # - report replaced command line entrypoints
      # - report docker run args
      return execution_result_pb2.ExecutorOutput()
  
file_path: tfx/tfx/orchestration/portable/docker_executor_operator.py
imports: from typing import Any, Dict, Optional, cast
  from absl import logging
  import docker
  from tfx.dsl.compiler import placeholder_utils
  from tfx.dsl.component.experimental import executor_specs
  from tfx.orchestration.portable import base_executor_operator
  from tfx.orchestration.portable import data_types
  from tfx.proto.orchestration import executable_spec_pb2
  from tfx.proto.orchestration import execution_result_pb2
  from tfx.proto.orchestration import platform_config_pb2
  from tfx.utils import proto_utils
  from google.protobuf import message
label_desc: Use image node_id 2 for calling the DockerExecutorOperator obj's underlying member method code with example usage: obj.run_executor(execution_info) and returns: execution_result_pb2