class_docstr: Executor for Kubernetes.
class_docstr_tok: ["Executor","for","Kubernetes","."]
docstr: Clear tasks that were not yet launched, but were previously queued.
  
  Tasks can end up in a "Queued" state when a rescheduled/deferred operator
  comes back up for execution (with the same try_number) before the
  pod of its previous incarnation has been fully removed (we think).
  
  It's also possible when an executor abruptly shuts down (leaving a non-empty
  task_queue on that executor), but that scenario is handled via normal adoption.
  
  This method checks each of our queued tasks to see if the corresponding pod
  is around, and if not, and there's no matching entry in our own
  task_queue, marks it for re-execution.
docstr_tok: ["Clear","tasks","that","were","not","yet","launched",",","but","were","previously","queued",".","Tasks","can","end","up","in","a","``","Queued","''","state","when","a","rescheduled\/deferred","operator","comes","back","up","for","execution","(","with","the","same","try_number",")","before","the","pod","of","its","previous","incarnation","has","been","fully","removed","(","we","think",")",".","It","'s","also","possible","when","an","executor","abruptly","shuts","down","(","leaving","a","non-empty","task_queue","on","that","executor",")",",","but","that","scenario","is","handled","via","normal","adoption",".","This","method","checks","each","of","our","queued","tasks","to","see","if","the","corresponding","pod","is","around",",","and","if","not",",","and","there","'s","no","matching","entry","in","our","own","task_queue",",","marks","it","for","re-execution","."]
code: def clear_not_launched_queued_tasks(
      self, session: Session = NEW_SESSION
  ) -> None:
      """
      Clear tasks that were not yet launched, but were previously queued.
  
      Tasks can end up in a "Queued" state when a rescheduled/deferred operator
      comes back up for execution (with the same try_number) before the
      pod of its previous incarnation has been fully removed (we think).
  
      It's also possible when an executor abruptly shuts down (leaving a non-empty
      task_queue on that executor), but that scenario is handled via normal adoption.
  
      This method checks each of our queued tasks to see if the corresponding pod
      is around, and if not, and there's no matching entry in our own
      task_queue, marks it for re-execution.
      """
      if TYPE_CHECKING:
          assert self.kube_client
      from airflow.models.taskinstance import TaskInstance
  
      with Stats.timer(
          "kubernetes_executor.clear_not_launched_queued_tasks.duration"
      ):
          self.log.debug("Clearing tasks that have not been launched")
          query = select(TaskInstance).where(
              TaskInstance.state == TaskInstanceState.QUEUED,
              TaskInstance.queued_by_job_id == self.job_id,
          )
          if self.kubernetes_queue:
              query = query.where(
                  TaskInstance.queue == self.kubernetes_queue
              )
          queued_tis: list[TaskInstance] = session.scalars(query).all()
          self.log.info(
              "Found %s queued task instances", len(queued_tis)
          )
  
          # Go through the "last seen" dictionary and clean out old entries
          allowed_age = (
              self.kube_config.worker_pods_queued_check_interval * 3
          )
          for key, timestamp in list(self.last_handled.items()):
              if time.time() - timestamp > allowed_age:
                  del self.last_handled[key]
  
          if not queued_tis:
              return
  
          # airflow worker label selector batch call
          kwargs = {
              "label_selector": f"airflow-worker={self._make_safe_label_value(str(self.job_id))}"
          }
          if self.kube_config.kube_client_request_args:
              kwargs.update(self.kube_config.kube_client_request_args)
          pod_list = self._list_pods(kwargs)
  
          # create a set against pod query label fields
          label_search_set = set()
          for pod in pod_list:
              dag_id = pod.metadata.labels.get("dag_id", None)
              task_id = pod.metadata.labels.get("task_id", None)
              airflow_worker = pod.metadata.labels.get(
                  "airflow-worker", None
              )
              map_index = pod.metadata.labels.get("map_index", None)
              run_id = pod.metadata.labels.get("run_id", None)
              execution_date = pod.metadata.labels.get(
                  "execution_date", None
              )
              if (
                  dag_id is None
                  or task_id is None
                  or airflow_worker is None
              ):
                  continue
              label_search_base_str = f"dag_id={dag_id},task_id={task_id},airflow-worker={airflow_worker}"
              if map_index is not None:
                  label_search_base_str += f",map_index={map_index}"
              if run_id is not None:
                  label_search_str = (
                      f"{label_search_base_str},run_id={run_id}"
                  )
                  label_search_set.add(label_search_str)
              if execution_date is not None:
                  label_search_str = f"{label_search_base_str},execution_date={execution_date}"
                  label_search_set.add(label_search_str)
  
          for ti in queued_tis:
              self.log.debug("Checking task instance %s", ti)
  
              # Check to see if we've handled it ourselves recently
              if ti.key in self.last_handled:
                  continue
  
              # Build the pod selector
              base_label_selector = (
                  f"dag_id={self._make_safe_label_value(ti.dag_id)},"
                  f"task_id={self._make_safe_label_value(ti.task_id)},"
                  f"airflow-worker={self._make_safe_label_value(str(ti.queued_by_job_id))}"
              )
              if ti.map_index >= 0:
                  # Old tasks _couldn't_ be mapped, so we don't have to worry about compat
                  base_label_selector += f",map_index={ti.map_index}"
  
              # Try run_id first
              label_search_str = f"{base_label_selector},run_id={self._make_safe_label_value(ti.run_id)}"
              if label_search_str in label_search_set:
                  continue
              # Fallback to old style of using execution_date
              label_search_str = f"{base_label_selector},execution_date={self._make_safe_label_value(ti.execution_date)}"
              if label_search_str in label_search_set:
                  continue
              self.log.info(
                  "TaskInstance: %s found in queued state but was not launched, rescheduling",
                  ti,
              )
              session.execute(
                  update(TaskInstance)
                  .where(
                      TaskInstance.dag_id == ti.dag_id,
                      TaskInstance.task_id == ti.task_id,
                      TaskInstance.run_id == ti.run_id,
                      TaskInstance.map_index == ti.map_index,
                  )
                  .values(state=TaskInstanceState.SCHEDULED)
              )
  
imports: from __future__ import annotations
  import contextlib
  import json
  import logging
  import multiprocessing
  import time
  from collections import defaultdict
  from contextlib import suppress
  from datetime import datetime
  from queue import Empty, Queue
  from typing import TYPE_CHECKING, Any, Sequence
  from kubernetes.dynamic import DynamicClient
  from sqlalchemy import select, update
  from airflow.providers.cncf.kubernetes.pod_generator import PodMutationHookException, PodReconciliationError
  from airflow.stats import Stats
  from airflow.configuration import conf
  from airflow.executors.base_executor import BaseExecutor
  from airflow.providers.cncf.kubernetes.executors.kubernetes_executor_types import ADOPTED, POD_EXECUTOR_DONE_KEY
  from airflow.providers.cncf.kubernetes.kube_config import KubeConfig
  from airflow.providers.cncf.kubernetes.kubernetes_helper_functions import annotations_to_key
  from airflow.utils.event_scheduler import EventScheduler
  from airflow.utils.log.logging_mixin import remove_escape_codes
  from airflow.utils.session import NEW_SESSION, provide_session
  from airflow.utils.state import TaskInstanceState
label_desc: Use image node_id 4 for calling the KubernetesExecutor obj's underlying member method code with example usage: obj.clear_not_launched_queued_tasks(session) without return types