Currently, the celery executor imports the `Task` model for the Celery database backend directly:

https://github.com/apache/airflow/blob/2dd4e96045d4a7f45cc8c06df3d25c4f1479392c/airflow/executors/celery_executor.py#L38

However, Celery, being highly customisable, uses `self.task_cls` in the backend implementation; this defaults to either `Task` or `TaskExtended` depending on the Celery configuration, see [`celery.backends.database:66`](https://github.com/celery/celery/blob/406f04a082949ac42ec7a4af94fed896c515aaa4/celery/backends/database/__init__.py#L66):

> ```python
>     task_cls = Task
> ```

and [`celery.backends.database:76,77`](https://github.com/celery/celery/blob/406f04a082949ac42ec7a4af94fed896c515aaa4/celery/backends/database/__init__.py#L76-L77):

> ```python
>         if self.extended_result:
>             self.task_cls = TaskExtended
> ```

Airflow should support this flexibility, and instead of importing a fixed class just use `backend.task_cls`, so in `_get_many_from_db_backend` just use:

```python
        Task = app.backend.task_cls
        with session_cleanup(session):
            tasks = session.query(Task).filter(Task.task_id.in_(task_ids)).all()
```

I suppose there is one downside to this: `TaskExtended` fetches a few more columns per task, but given that the result of `task.to_dict()` is passed to `backend.meta_from_decoded()`, which could also have been customised to make use of information contained in a custom model, that's neither here nor there. It would be the price anyone using the `result_extended` option would need to pay elsewhere anyway.
airflow/executors/celery_executor.py
"""
        See if any of the tasks we adopted from another Executor run have not
        progressed after the configured timeout.
        If they haven't, they likely never made it to Celery, and we should
        just resend them. We do that by clearing the state and letting the
        normal scheduler loop deal with that
        """
        now = utcnow()
        timedout_keys = []
        for key, stalled_after in self.adopted_task_timeouts.items():
            if stalled_after > now:
                
                
                break
            
            
            
            
            
            
            
            timedout_keys.append(key)
        if timedout_keys:
            self.log.error(