**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
from celery.result import AsyncResult
from celery.signals import import_modules as celery_import_modules
from setproctitle import setproctitle
import airflow.settings as settings
from airflow.config_templates.default_celery import DEFAULT_CELERY_CONFIG
from airflow.configuration import conf
from airflow.exceptions import AirflowException, AirflowTaskTimeout
from airflow.executors.base_executor import BaseExecutor, CommandType, EventBufferValueType
from airflow.models.taskinstance import TaskInstance, TaskInstanceKey
from airflow.stats import Stats
from airflow.utils.log.logging_mixin import LoggingMixin
from airflow.utils.net import get_hostname
from airflow.utils.state import State
from airflow.utils.timeout import timeout
from airflow.utils.timezone import utcnow
log = logging.getLogger(__name__)
CELERY_FETCH_ERR_MSG_HEADER = 'Error fetching Celery task state'
CELERY_SEND_ERR_MSG_HEADER = 'Error sending Celery task'
OPERATION_TIMEOUT = conf.getfloat('celery', 'operation_timeout', fallback=1.0)
'''
To start the celery worker, run the command:
airflow celery worker
'''
if conf.has_option('celery', 'celery_config_options'):
    celery_configuration = conf.getimport('celery', 'celery_config_options')
else:
    celery_configuration = DEFAULT_CELERY_CONFIG
app = Celery(conf.get('celery', 'CELERY_APP_NAME'), config_source=celery_configuration)
@app.task
def execute_command(command_to_exec: CommandType) -> None: