docstr: Tests that a node previously cancelled due to stopping can be restarted.
docstr_tok: ["Tests","that","a","node","previously","cancelled","due","to","stopping","can","be","restarted","."]
code: def test_restart_node_cancelled_due_to_stopping(self):
      """Tests that a node previously cancelled due to stopping can be restarted."""
      test_utils.fake_example_gen_run(
          self._mlmd_connection, self._example_gen, 1, 1
      )
  
      [stats_gen_task] = self._generate_and_test(
          False,
          num_initial_executions=1,
          num_tasks_generated=1,
          num_new_executions=1,
          num_active_executions=1,
          ignore_update_node_state_tasks=True,
      )
      node_uid = task_lib.NodeUid.from_node(
          self._pipeline, self._stats_gen
      )
      self.assertEqual(node_uid, stats_gen_task.node_uid)
  
      # Simulate stopping the node while it is under execution, which leads to
      # the node execution being cancelled.
      with self._mlmd_connection as m:
          with mlmd_state.mlmd_execution_atomic_op(
              m, stats_gen_task.execution_id
          ) as stats_gen_exec:
              stats_gen_exec.last_known_state = (
                  metadata_store_pb2.Execution.CANCELED
              )
              data_types_utils.set_metadata_value(
                  stats_gen_exec.custom_properties[
                      constants.EXECUTION_ERROR_MSG_KEY
                  ],
                  "manually stopped",
              )
  
      # 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)
  
      # New execution should be created for any previously canceled node when the
      # node state is STARTING.
      [
          update_node_state_task,
          stats_gen_task,
      ] = self._generate_and_test(
          False,
          num_initial_executions=2,
          num_tasks_generated=2,
          num_new_executions=1,
          num_active_executions=1,
      )
      self.assertIsInstance(
          update_node_state_task, task_lib.UpdateNodeStateTask
      )
      self.assertEqual(node_uid, update_node_state_task.node_uid)
      self.assertEqual(
          pstate.NodeState.RUNNING, update_node_state_task.state
      )
      self.assertEqual(node_uid, stats_gen_task.node_uid)
  
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 19 for calling the SyncPipelineTaskGeneratorTest obj's underlying member method code with example usage: obj.test_restart_node_cancelled_due_to_stopping() without return types