**Apache Airflow version**: 2.1.1

**What happened**:

`_check_for_stalled_adopted_tasks` method breaks on first item which satisfies condition
https://github.com/apache/airflow/blob/2.1.1/airflow/executors/celery_executor.py#L353

From the comment/logic, it looks like the idea is to optimize this piece of code, however it is not seen that `self.adopted_task_timeouts` object maintains entries sorted by timestamp. This results in unstable behaviour of scheduler, which means it sometimes may not resend tasks to Celery (due to skipping them).

Confirmed with Airflow 2.1.1

**What you expected to happen**:

Deterministic behaviour of scheduler in this case

**How to reproduce it**:

These are steps to reproduce adoption of tasks. To reproduce unstable behaviour, you may need to do trigger some additional DAGs in the process.

- set [core]parallelism to 30
- trigger DAG with concurrency==100 and 30 tasks, each is running for 30 minutes (e.g. sleep 1800)
- 18 of them will be running, others will be in queued state
- restart scheduler
- observe "Adopted tasks were still pending after 0:10:00, assuming they never made it to celery and clearing:"
- tasks will be failed and marked as "up for retry"

The important thing is that scheduler has to restart once tasks get to the queue, so that it will adopt queued tasks.
airflow/executors/celery_executor.py
"""
    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.
    """
    def __init__(self):
        super().__init__()
        
        
        
        self._sync_parallelism = conf.getint('celery', 'SYNC_PARALLELISM')
        if self._sync_parallelism == 0:
            self._sync_parallelism = max(1, cpu_count() - 1)
        self.bulk_state_fetcher = BulkStateFetcher(self._sync_parallelism)
        self.tasks = {}
        
        self.adopted_task_timeouts: Dict[TaskInstanceKey, datetime.datetime] = OrderedDict()
        self.task_adoption_timeout = datetime.timedelta(
            seconds=conf.getint('celery', 'task_adoption_timeout', fallback=600)
        )
        self.task_publish_retries: Dict[TaskInstanceKey, int] = OrderedDict()
        self.task_publish_max_retries = conf.getint('celery', 'task_publish_max_retries', fallback=3)
    def start(self) -> None: