class_docstr: An end to end test using fully orchestrated Airflow.
class_docstr_tok: ["An","end","to","end","test","using","fully","orchestrated","Airflow","."]
code: def _CheckPendingTasks(
      self, pending_task_names: Sequence[str]
  ) -> Set[str]:
      unknown_tasks = set(pending_task_names) - set(self._all_tasks)
      assert not unknown_tasks, "Unknown task name {}".format(
          unknown_tasks
      )
      still_pending = set()
      failed = dict()
      for task in pending_task_names:
          task_state = self._GetState(task).lower()
          if task_state in _SUCCESS_TASK_STATES:
              absl.logging.info(
                  "Task %s succeeded, checking output artifacts", task
              )
              self._CheckOutputArtifacts(task)
          elif task_state in _PENDING_TASK_STATES:
              still_pending.add(task)
          else:
              failed[task] = task_state
      for task, state in failed.items():
          absl.logging.error(
              "Retrieving logs for %s task %s", state, task
          )
          self._PrintTaskLogsOnError(task)
      self.assertFalse(failed)
      return still_pending
  
imports: import os
  import platform
  import subprocess
  import time
  from typing import Sequence, Set
  import unittest
  import absl
  import tensorflow
  from tfx.dsl.io import fileio
  from tfx.orchestration.airflow import test_utils
  from tfx.tools.cli.e2e import test_utils
  from tfx.utils import io_utils
  from tfx.utils import test_case_utils
label_desc: Use image node_id 4 for calling the AirflowEndToEndTest obj's underlying member method code with example usage: obj._CheckPendingTasks(pending_task_names) and returns: still_pending