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_graceful_handling_if_error_publishing_scheduler_results(
      self, mock_publish
  ):
      def _publish(**kwargs):
          raise ValueError("test error")
  
      mock_publish.side_effect = _publish
  
      # Register a fake task scheduler that returns a successful execution result
      # and `OK` task scheduler status.
      self._register_task_scheduler(
          ts.TaskSchedulerResult(
              status=status_lib.Status(code=status_lib.Code.OK),
              output=ts.ImporterNodeOutput(
                  output_artifacts=self._task.output_artifacts
              ),
          )
      )
  
      task_manager = self._run_task_manager()
      mock_publish.assert_called_once()
      self.assertTrue(task_manager.done())
      self.assertIsNone(task_manager.exception())
  
      # Verify that execution is marked as failed.
      execution = self._get_execution()
      self.assertEqual(
          metadata_store_pb2.Execution.FAILED,
          execution.last_known_state,
      )
      self.assertEqual(
          "test error",
          data_types_utils.get_metadata_value(
              execution.custom_properties[
                  constants.EXECUTION_ERROR_MSG_KEY
              ]
          ),
      )
  
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 11 for calling the TaskManagerE2ETest obj's underlying member method code with example usage: obj.test_graceful_handling_if_error_publishing_scheduler_results(mock_publish) without return types