Attempt to stabilize flaky celery integration test (apache#39892)

tests/integration/executors/test_celery_executor.py

@@ -23,6 +23,7 @@
 import os
 import sys
 from datetime import datetime
+from time import sleep
 from unittest import mock
 
 # leave this it is used by the test worker
@@ -42,6 +43,8 @@
 from airflow.utils.state import State
 from tests.test_utils import db
 
+logger = logging.getLogger(__name__)
+
 
 def _prepare_test_bodies():
     if "CELERY_BROKER_URLS" in os.environ:
@@ -145,7 +148,15 @@ def fake_execute_command(command):
                     executor.task_publish_retries[key] = 1
 
                 executor._process_tasks(task_tuples_to_send)
-
+                for _ in range(20):
+                    num_tasks = len(executor.tasks.keys())
+                    if num_tasks == 2:
+                        break
+                    logger.info(
+                        "Waiting 0.1 s for tasks to be processed asynchronously. Processed so far %d",
+                        num_tasks,
+                    )
+                    sleep(0.4)
                 assert list(executor.tasks.keys()) == [
                     ("success", "fake_simple_ti", execute_date, 0),
                     ("fail", "fake_simple_ti", execute_date, 0),