docstr: Tests component retry when node is STARTING.
docstr_tok: ["Tests","component","retry","when","node","is","STARTING","."]
code: def test_component_retry_when_node_is_starting(self):
      """Tests component retry when node is STARTING."""
      test_utils.fake_example_gen_run(
          self._mlmd_connection, self._example_gen, 1, 1
      )
      node_uid = task_lib.NodeUid.from_node(
          self._pipeline, self._stats_gen
      )
  
      self._stats_gen.execution_options.max_execution_retries = 2
      [exec_node_task] = self._generate(False, True, fail_fast=True)
      self.assertEqual(
          self._stats_gen.node_info.id, exec_node_task.node_uid.node_id
      )
  
      # Simulate fail and rerun StatsGen twice.
      for _ in range(
          self._stats_gen.execution_options.max_execution_retries
      ):
          # Simulate StatsGen failure.
          with self._mlmd_connection as m:
              with mlmd_state.mlmd_execution_atomic_op(
                  m, exec_node_task.execution_id
              ) as ev_exec:
                  ev_exec.last_known_state = (
                      metadata_store_pb2.Execution.FAILED
                  )
  
          # It should generate a ExecNodeTask due to retry.
          [update_node_task, exec_node_task] = self._generate(
              False, False, fail_fast=True
          )
          self.assertIsInstance(exec_node_task, task_lib.ExecNodeTask)
          self.assertEqual(
              self._stats_gen.node_info.id,
              exec_node_task.node_uid.node_id,
          )
          self.assertIsInstance(
              update_node_task, task_lib.UpdateNodeStateTask
          )
          self.assertEqual(
              update_node_task.state, pstate.NodeState.RUNNING
          )
  
      # Fail StatsGen the third time.
      with self._mlmd_connection as m:
          with mlmd_state.mlmd_execution_atomic_op(
              m, exec_node_task.execution_id
          ) as ev_exec:
              ev_exec.last_known_state = (
                  metadata_store_pb2.Execution.FAILED
              )
  
      # Change state of node to STARTING.
      with self._mlmd_connection as m:
          pipeline_state = test_utils.get_or_create_pipeline_state(
              m, self._pipeline
          )
          with pipeline_state:
              with pipeline_state.node_state_update_context(
                  node_uid
              ) as node_state:
                  node_state.update(pstate.NodeState.STARTING)
  
      # It should generate a ExecNodeTask due to state being STARTING.
      [update_node_task, exec_node_task] = self._generate(
          False, False, fail_fast=True
      )
      self.assertIsInstance(exec_node_task, task_lib.ExecNodeTask)
      self.assertEqual(
          self._stats_gen.node_info.id, exec_node_task.node_uid.node_id
      )
      self.assertIsInstance(
          update_node_task, task_lib.UpdateNodeStateTask
      )
      self.assertEqual(update_node_task.state, pstate.NodeState.RUNNING)
  
imports: import itertools
  import os
  from typing import Literal
  import uuid
  from absl.testing import parameterized
  from absl.testing.absltest import mock
  import tensorflow
  from tfx.dsl.compiler import constants
  from tfx.orchestration import data_types_utils
  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 service_jobs
  from tfx.orchestration.experimental.core import sync_pipeline_task_gen
  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_sync_pipeline
  from tfx.orchestration import mlmd_connection_manager
  from tfx.orchestration.portable import runtime_parameter_utils
  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 25 for calling the SyncPipelineTaskGeneratorTest obj's underlying member method code with example usage: obj.test_component_retry_when_node_is_starting() without return types