class_docstr: CeleryExecutor is recommended for production use of Airflow.
  
  It allows distributing the execution of task instances to multiple worker nodes.
  
  Celery is a simple, flexible and reliable distributed system to process
  vast amounts of messages, while providing operations with the tools
  required to maintain such a system.
class_docstr_tok: ["CeleryExecutor","is","recommended","for","production","use","of","Airflow",".","It","allows","distributing","the","execution","of","task","instances","to","multiple","worker","nodes",".","Celery","is","a","simple",",","flexible","and","reliable","distributed","system","to","process","vast","amounts","of","messages",",","while","providing","operations","with","the","tools","required","to","maintain","such","a","system","."]
code: def try_adopt_task_instances(
      self, tis: Sequence[TaskInstance]
  ) -> Sequence[TaskInstance]:
      # See which of the TIs are still alive (or have finished even!)
      #
      # Since Celery doesn't store "SENT" state for queued commands (if we create an AsyncResult with a made
      # up id it just returns PENDING state for it), we have to store Celery's task_id against the TI row to
      # look at in future.
      #
      # This process is not perfect -- we could have sent the task to celery, and crashed before we were
      # able to record the AsyncResult.task_id in the TaskInstance table, in which case we won't adopt the
      # task (it'll either run and update the TI state, or the scheduler will clear and re-queue it. Either
      # way it won't get executed more than once)
      #
      # (If we swapped it around, and generated a task_id for Celery, stored that in TI and enqueued that
      # there is also still a race condition where we could generate and store the task_id, but die before
      # we managed to enqueue the command. Since neither way is perfect we always have to deal with this
      # process not being perfect.)
      from celery.result import AsyncResult
  
      celery_tasks = {}
      not_adopted_tis = []
  
      for ti in tis:
          if ti.external_executor_id is not None:
              celery_tasks[ti.external_executor_id] = (
                  AsyncResult(ti.external_executor_id),
                  ti,
              )
          else:
              not_adopted_tis.append(ti)
  
      if not celery_tasks:
          # Nothing to adopt
          return tis
  
      states_by_celery_task_id = self.bulk_state_fetcher.get_many(
          list(map(operator.itemgetter(0), celery_tasks.values()))
      )
  
      adopted = []
      cached_celery_backend = next(iter(celery_tasks.values()))[
          0
      ].backend
  
      for celery_task_id, (
          state,
          info,
      ) in states_by_celery_task_id.items():
          result, ti = celery_tasks[celery_task_id]
          result.backend = cached_celery_backend
  
          # Set the correct elements of the state dicts, then update this
          # like we just queried it.
          self.tasks[ti.key] = result
          self.running.add(ti.key)
          self.update_task_state(ti.key, state, info)
          adopted.append(f"{ti} in state {state}")
  
      if adopted:
          task_instance_str = "\n\t".join(adopted)
          self.log.info(
              "Adopted the following %d tasks from a dead executor\n\t%s",
              len(adopted),
              task_instance_str,
          )
  
      return not_adopted_tis
  
imports: from __future__ import annotations
  import logging
  import math
  import operator
  import time
  from collections import Counter
  from concurrent.futures import ProcessPoolExecutor
  from multiprocessing import cpu_count
  from typing import TYPE_CHECKING, Any, Optional, Sequence, Tuple
  from celery import states
  from airflow.configuration import conf
  from airflow.exceptions import AirflowTaskTimeout
  from airflow.executors.base_executor import BaseExecutor
  from airflow.stats import Stats
  from airflow.utils.state import TaskInstanceState
label_desc: Use image node_id 13 for calling the CeleryExecutor obj's underlying member method code with example usage: obj.try_adopt_task_instances(tis) and returns: not_adopted_tis, tis