docstr: Tests task generation when example-gen service job fails.
docstr_tok: ["Tests","task","generation","when","example-gen","service","job","fails","."]
code: def test_service_job_failed(self):
      """Tests task generation when example-gen service job fails."""
  
      def _ensure_node_services(
          unused_pipeline_state, node_id, unused_backfill_token=""
      ):
          if node_id == "my_example_gen":
              return service_jobs.ServiceStatus(
                  code=service_jobs.ServiceStatusCode.FAILED,
                  msg="foobar error",
              )
  
      self._mock_service_job_manager.ensure_node_services.side_effect = (
          _ensure_node_services
      )
      [
          update_examplegen,
          update_transform,
          update_trainer,
      ] = self._generate_and_test(
          True,
          num_initial_executions=0,
          num_tasks_generated=3,
          num_new_executions=0,
          num_active_executions=0,
      )
      self.assertIsInstance(
          update_examplegen, task_lib.UpdateNodeStateTask
      )
      self.assertEqual(
          status_lib.Code.UNKNOWN, update_examplegen.status.code
      )
      self.assertEqual(
          "associated service job failed; node uid:"
          " NodeUid(pipeline_uid=PipelineUid(pipeline_id='my_pipeline',"
          " pipeline_run_id=None), node_id='my_example_gen'); error message:"
          " foobar error",
          update_examplegen.status.message,
      )
      self.assertIsInstance(
          update_transform, task_lib.UpdateNodeStateTask
      )
      self.assertEqual(status_lib.Code.OK, update_transform.status.code)
      self.assertEqual(
          "Waiting for new input artifacts to be processed. Non-triggering input"
          " or insufficient number of artifacts will not trigger new execution.",
          update_transform.status.message,
      )
      self.assertIsInstance(
          update_trainer, task_lib.UpdateNodeStateTask
      )
      self.assertEqual(status_lib.Code.OK, update_trainer.status.code)
      self.assertEqual(
          "Waiting for new input artifacts to be processed. Non-triggering input"
          " or insufficient number of artifacts will not trigger new execution.",
          update_trainer.status.message,
      )
  
imports: import os
  from absl.testing import parameterized
  from absl.testing.absltest import mock
  import tensorflow
  from tfx.orchestration import node_proto_view
  from tfx.orchestration.experimental.core import async_pipeline_task_gen
  from tfx.orchestration.experimental.core import pipeline_state
  from tfx.orchestration.experimental.core import service_jobs
  from tfx.orchestration.experimental.core import task
  from tfx.orchestration.experimental.core import task_gen_utils
  from tfx.orchestration.experimental.core import task_queue
  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.orchestration.portable.input_resolution import exceptions
  from tfx.utils import status
label_desc: Use image node_id 9 for calling the AsyncPipelineTaskGeneratorTest obj's underlying member method code with example usage: obj.test_service_job_failed() and returns: service_jobs