class_docstr: Test end-to-end from task generation to publication of results to MLMD.
class_docstr_tok: ["Test","end-to-end","from","task","generation","to","publication","of","results","to","MLMD","."]
code: def test_scheduler_failure(self):
      # Register a fake task scheduler that returns a failure status.
      self._register_task_scheduler(
          ts.TaskSchedulerResult(
              status=status_lib.Status(
                  code=status_lib.Code.ABORTED, message="foobar error"
              )
          )
      )
      task_manager = self._run_task_manager()
      self.assertTrue(task_manager.done())
      self.assertIsNone(task_manager.exception())
  
      # Check that the task was processed and MLMD execution marked failed.
      self.assertTrue(self._task_queue.is_empty())
      execution = self._get_execution()
      self.assertEqual(
          metadata_store_pb2.Execution.FAILED,
          execution.last_known_state,
      )
      self.assertEqual(
          "foobar error",
          data_types_utils.get_metadata_value(
              execution.custom_properties[
                  constants.EXECUTION_ERROR_MSG_KEY
              ]
          ),
      )
  
      # Check that stateful working dir still exists, but tmp_dir is removed.
      self.assertTrue(os.path.exists(self._task.stateful_working_dir))
      self.assertFalse(os.path.exists(self._task.tmp_dir))
  
imports: import contextlib
  import functools
  import os
  import threading
  import time
  from absl import logging
  from absl.testing.absltest import mock
  import tensorflow
  from tfx.orchestration import data_types_utils
  from tfx.orchestration import metadata
  from tfx.orchestration.experimental.core import async_pipeline_task_gen
  from tfx.orchestration.experimental.core import constants
  from tfx.orchestration.experimental.core import pipeline_state
  from tfx.orchestration.experimental.core import post_execution_utils
  from tfx.orchestration.experimental.core import service_jobs
  from tfx.orchestration.experimental.core import task
  from tfx.orchestration.experimental.core import task_manager
  from tfx.orchestration.experimental.core import task_queue
  from tfx.orchestration.experimental.core import task_scheduler
  from tfx.orchestration.experimental.core import test_utils
  from tfx.orchestration.experimental.core.testing import test_async_pipeline
  from tfx.orchestration import mlmd_connection_manager
  from tfx.proto.orchestration import execution_result_pb2
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.utils import status
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 7 for calling the TaskManagerE2ETest obj's underlying member method code with example usage: obj.test_scheduler_failure() without return types