class_docstr: TaskManager acts on the tasks fetched from the task queues.
  
  TaskManager instance can be used as a context manager:
class_docstr_tok: ["TaskManager","acts","on","the","tasks","fetched","from","the","task","queues",".","TaskManager","instance","can","be","used","as","a","context","manager",":"]
docstr: Processes an `ExecNodeTask` using the given task scheduler.
docstr_tok: ["Processes","an","`","ExecNodeTask","`","using","the","given","task","scheduler","."]
code: def _process_exec_node_task(
      self, scheduler: _SchedulerWrapper, task: task_lib.ExecNodeTask
  ) -> None:
      """Processes an `ExecNodeTask` using the given task scheduler."""
      # This is a blocking call to the scheduler which can take a long time to
      # complete for some types of task schedulers. The scheduler is expected to
      # handle any internal errors gracefully and return the result with an error
      # status. But in case the scheduler raises an exception, it is considered
      # a failed execution and MLMD is updated accordingly.
      try:
          result = scheduler.schedule()
      except Exception as e:  # pylint: disable=broad-except
          logging.exception("Exception raised by: %s", scheduler)
          if isinstance(e, status_lib.StatusNotOkError):
              status = status_lib.Status(code=e.code, message=e.message)
          else:
              status = status_lib.Status(
                  code=status_lib.Code.UNKNOWN,
                  message=str(sys.exc_info()),
              )
          result = ts.TaskSchedulerResult(status=status)
      logging.info(
          "TaskSchedulerResult status %s from running %s",
          result.status,
          scheduler,
      )
      # If the node was paused, we do not complete the execution as it is expected
      # that a new ExecNodeTask would be issued for resuming the execution.
      if not (
          scheduler.pause_requested.is_set()
          and result.status.code == status_lib.Code.CANCELLED
      ):
          try:
              post_execution_utils.publish_execution_results_for_task(
                  mlmd_handle=self._mlmd_handle,
                  task=task,
                  result=result,
              )
          except Exception as e:  # pylint: disable=broad-except
              logging.exception(
                  (
                      "Attempting to mark execution (id: %s) as FAILED after failure"
                      " to publish task scheduler execution results: %s"
                  ),
                  task.execution_id,
                  result,
              )
              self._fail_execution(
                  task.execution_id, status_lib.Code.UNKNOWN, str(e)
              )
          pipeline_state.record_state_change_time()
      with self._tm_lock:
          del self._scheduler_by_node_uid[task.node_uid]
          self._task_queue.task_done(task)
  
imports: from concurrent import futures
  import datetime
  import sys
  import textwrap
  import threading
  import time
  import typing
  from typing import Dict, List, Optional
  from absl import logging
  import pytz
  from tfx.orchestration import data_types_utils
  from tfx.orchestration import metadata
  from tfx.orchestration.experimental.core import constants
  from tfx.orchestration.experimental.core import mlmd_state
  from tfx.orchestration.experimental.core import pipeline_state
  from tfx.orchestration.experimental.core import post_execution_utils
  from tfx.orchestration.experimental.core import task
  from tfx.orchestration.experimental.core import task_queue
  from tfx.orchestration.experimental.core import task_scheduler
  from tfx.utils import status
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 10 for calling the TaskManager obj's underlying member method code with example usage: obj._process_exec_node_task(scheduler, task) without return types