code: def test_manual_task_scheduler(self):
      task_queue = tq.TaskQueue()
  
      [manual_task] = test_utils.run_generator_and_test(
          test_case=self,
          mlmd_connection_manager=self._mlmd_cm,
          generator_class=sptg.SyncPipelineTaskGenerator,
          pipeline=self._pipeline,
          task_queue=task_queue,
          use_task_queue=True,
          service_job_manager=None,
          num_initial_executions=0,
          num_tasks_generated=1,
          num_new_executions=1,
          num_active_executions=1,
          expected_exec_nodes=[self._manual_node],
          ignore_update_node_state_tasks=True,
      )
  
      ts_result = []
  
      def start_scheduler(ts_result):
          with self._mlmd_connection as m:
              ts_result.append(
                  manual_task_scheduler.ManualTaskScheduler(
                      mlmd_handle=m,
                      pipeline=self._pipeline,
                      task=manual_task,
                  ).schedule()
              )
  
      # Marks the execution as COMPLETE.
      def resume_node():
          task = typing.cast(task_lib.ExecNodeTask, manual_task)
          with mlmd_state.mlmd_execution_atomic_op(
              mlmd_handle=self._mlmd_connection,
              execution_id=task.execution_id,
          ) as execution:
              completed_state = manual_task_scheduler.ManualNodeState(
                  state=manual_task_scheduler.ManualNodeState.COMPLETED
              )
              completed_state.set_mlmd_value(
                  execution.custom_properties.get_or_create(
                      manual_task_scheduler.NODE_STATE_PROPERTY_KEY
                  )
              )
  
      # Shortens the polling interval during test.
      manual_task_scheduler._POLLING_INTERVAL_SECS = 1
  
      # Starts task scheduler and keeps polling for the node state.
      # The scheduler should be blocked (ts_result has nothing)
      # because the node state stays in WAITING.
      threading.Thread(
          target=start_scheduler, args=(ts_result,)
      ).start()
      self.assertEqual(len(ts_result), 0)
      time.sleep(manual_task_scheduler._POLLING_INTERVAL_SECS * 10)
      self.assertEqual(len(ts_result), 0)
  
      # Changes node state to COMPLETED in another thread.
      threading.Thread(target=resume_node).start()
      # Waits for the state change to propagate through.
      time.sleep(manual_task_scheduler._POLLING_INTERVAL_SECS * 10)
      self.assertEqual(len(ts_result), 1)
      self.assertEqual(status_lib.Code.OK, ts_result[0].status.code)
      self.assertIsInstance(ts_result[0].output, ts.ExecutorNodeOutput)
  
imports: import os
  import threading
  import time
  import typing
  import uuid
  import tensorflow
  from tfx.dsl.compiler import constants
  from tfx.orchestration.experimental.core import mlmd_state
  from tfx.orchestration.experimental.core import sync_pipeline_task_gen
  from tfx.orchestration.experimental.core import task
  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.task_schedulers import manual_task_scheduler
  from tfx.orchestration.experimental.core.testing import test_manual_node
  from tfx.orchestration import mlmd_connection_manager
  from tfx.orchestration.portable import runtime_parameter_utils
  from tfx.utils import status
label_desc: Use image node_id 3 for calling the ManualTaskSchedulerTest obj's underlying member method code with example usage: obj.test_manual_task_scheduler() without return types