class_docstr: Executor for Kubernetes.
class_docstr_tok: ["Executor","for","Kubernetes","."]
code: def try_adopt_task_instances(
      self, tis: Sequence[TaskInstance]
  ) -> Sequence[TaskInstance]:
      with Stats.timer(
          "kubernetes_executor.adopt_task_instances.duration"
      ):
          # Always flush TIs without queued_by_job_id
          tis_to_flush = [ti for ti in tis if not ti.queued_by_job_id]
          scheduler_job_ids = {ti.queued_by_job_id for ti in tis}
          tis_to_flush_by_key = {
              ti.key: ti for ti in tis if ti.queued_by_job_id
          }
          kube_client: client.CoreV1Api = self.kube_client
          for scheduler_job_id in scheduler_job_ids:
              scheduler_job_id = self._make_safe_label_value(
                  str(scheduler_job_id)
              )
              # We will look for any pods owned by the no-longer-running scheduler,
              # but will exclude only successful pods, as those TIs will have a terminal state
              # and not be up for adoption!
              # Those workers that failed, however, are okay to adopt here as their TI will
              # still be in queued.
              query_kwargs = {
                  "field_selector": "status.phase!=Succeeded",
                  "label_selector": (
                      "kubernetes_executor=True,"
                      f"airflow-worker={scheduler_job_id},{POD_EXECUTOR_DONE_KEY}!=True"
                  ),
              }
              pod_list = self._list_pods(query_kwargs)
              for pod in pod_list:
                  self.adopt_launched_task(
                      kube_client, pod, tis_to_flush_by_key
                  )
          self._adopt_completed_pods(kube_client)
          tis_to_flush.extend(tis_to_flush_by_key.values())
          return tis_to_flush
  
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 11 for calling the KubernetesExecutor obj's underlying member method code with example usage: obj.try_adopt_task_instances(tis) and returns: tis_to_flush