class_docstr: A backfill job runner consists of a dag or subdag for a specific time range.
  
  It triggers a set of task instance runs, in the right order and lasts for
  as long as it takes for the set of task instance to be completed.
docstr: Compare task instances' states with that of the executor.
  
  Expands downstream mapped tasks when necessary.
  
  :param running: dict of key, task to verify
  :return: An iterable of expanded TaskInstance per MappedTask
docstr_tok: ["Compare","task","instances","'","states","with","that","of","the","executor",".","Expands","downstream","mapped","tasks","when","necessary",".",":","param","running",":","dict","of","key",",","task","to","verify",":","return",":","An","iterable","of","expanded","TaskInstance","per","MappedTask"]
code: def _manage_executor_state(
      self,
      running: Mapping[TaskInstanceKey, TaskInstance],
      session: Session,
  ) -> Iterator[
      tuple[AbstractOperator, str, Sequence[TaskInstance], int]
  ]:
      """
      Compare task instances' states with that of the executor.
  
      Expands downstream mapped tasks when necessary.
  
      :param running: dict of key, task to verify
      :return: An iterable of expanded TaskInstance per MappedTask
      """
      executor = self.job.executor
      # list of tuples (dag_id, task_id, execution_date, map_index) of running tasks in executor
      buffered_events = list(executor.get_event_buffer().items())
      running_tis_ids = [
          (key.dag_id, key.task_id, key.run_id, key.map_index)
          for key, _ in buffered_events
          if key in running
      ]
      # list of TaskInstance of running tasks in executor (refreshed from db in batch)
      refreshed_running_tis = session.scalars(
          select(TaskInstance).where(
              tuple_(
                  TaskInstance.dag_id,
                  TaskInstance.task_id,
                  TaskInstance.run_id,
                  TaskInstance.map_index,
              ).in_(running_tis_ids)
          )
      ).all()
      # dict of refreshed TaskInstance by key to easily find them
      running_dict = {
          (ti.dag_id, ti.task_id, ti.run_id, ti.map_index): ti
          for ti in refreshed_running_tis
      }
      need_refresh = False
  
      for key, value in buffered_events:
          state, info = value
          ti_key = (key.dag_id, key.task_id, key.run_id, key.map_index)
          if ti_key not in running_dict:
              self.log.warning(
                  "%s state %s not in running=%s",
                  key,
                  state,
                  running.values(),
              )
              continue
  
          ti = running_dict[ti_key]
          if need_refresh:
              ti.refresh_from_db(session=session)
  
          self.log.debug("Executor state: %s task %s", state, ti)
  
          if (
              state
              in (TaskInstanceState.FAILED, TaskInstanceState.SUCCESS)
              and ti.state in self.STATES_COUNT_AS_RUNNING
          ):
              msg = (
                  f"Executor reports task instance {ti} finished ({state}) although the task says its "
                  f"{ti.state}. Was the task killed externally? Info: {info}"
              )
              self.log.error(msg)
              ti.handle_failure(error=msg)
              continue
  
          def _iter_task_needing_expansion() -> (
              Iterator[AbstractOperator]
          ):
              from airflow.models.mappedoperator import AbstractOperator
  
              for node in self.dag.get_task(
                  ti.task_id, include_subdags=True
              ).iter_mapped_dependants():
                  if isinstance(node, AbstractOperator):
                      yield node
                  else:  # A (mapped) task group. All its children need expansion.
                      yield from node.iter_tasks()
  
          if ti.state not in self.STATES_COUNT_AS_RUNNING:
              # Don't use ti.task; if this task is mapped, that attribute
              # would hold the unmapped task. We need to original task here.
              for node in _iter_task_needing_expansion():
                  new_tis, num_mapped_tis = node.expand_mapped_task(
                      ti.run_id, session=session
                  )
                  yield node, ti.run_id, new_tis, num_mapped_tis
  
imports: from __future__ import annotations
  import time
  from typing import TYPE_CHECKING, Any, Iterable, Iterator, Mapping, Sequence
  import attr
  import pendulum
  from sqlalchemy import select, tuple_, update
  from sqlalchemy.exc import OperationalError
  from sqlalchemy.orm.session import make_transient
  from tabulate import tabulate
  from airflow import models
  from airflow.exceptions import AirflowException, BackfillUnfinished, DagConcurrencyLimitReached, NoAvailablePoolSlot, PoolNotFound, TaskConcurrencyLimitReached
  from airflow.executors.executor_loader import ExecutorLoader
  from airflow.jobs.base_job_runner import BaseJobRunner
  from airflow.jobs.job import Job, perform_heartbeat
  from airflow.models import DAG, DagPickle
  from airflow.models.dagrun import DagRun
  from airflow.models.taskinstance import TaskInstance
  from airflow.ti_deps.dep_context import DepContext
  from airflow.ti_deps.dependencies_deps import BACKFILL_QUEUED_DEPS
  from airflow.timetables.base import DagRunInfo
  from airflow.utils import helpers, timezone
  from airflow.utils.configuration import conf, tmp_configuration_copy
  from airflow.utils.log.logging_mixin import LoggingMixin
  from airflow.utils.session import NEW_SESSION, provide_session
  from airflow.utils.state import DagRunState, State, TaskInstanceState
  from airflow.utils.types import DagRunType
label_desc: Use image node_id 3 for calling the BackfillJobRunner obj's underlying member method code with example usage: obj._manage_executor_state(running, session) without return types