This is an automated email from the ASF dual-hosted git repository.
ashb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 3b188b904fb Pre-assign external_executor_id at queuing time to prevent
duplicate execution on scheduler crash (#65594)
3b188b904fb is described below
commit 3b188b904fbb9525e17173f6ef1f5e5d388a75fc
Author: Ash Berlin-Taylor <[email protected]>
AuthorDate: Thu Apr 23 09:18:12 2026 +0100
Pre-assign external_executor_id at queuing time to prevent duplicate
execution on scheduler crash (#65594)
When a scheduler crashes between dispatching a task to Celery and
processing the QUEUED event that persists `external_executor_id`, the
replacement scheduler cannot adopt the in-flight task. Without the
Celery task ID in the database, `try_adopt_task_instances` has no
`AsyncResult` to look up, so the task is reset and re-queued — causing
duplicate execution of an already-running task.
Fix this by generating `external_executor_id` via a DB-side UUID
function (`gen_random_uuid` on PostgreSQL, `UUID()` on MySQL, a
Python `uuid4` registered on SQLite) in the same bulk UPDATE that
sets state=QUEUED. The ID is committed atomically with the state
transition — no second write, no race window. RETURNING is used on
PostgreSQL and SQLite to read back the generated UUIDs without a
second round-trip; MySQL falls back to a SELECT.
The CeleryExecutor passes the pre-assigned ID to `apply_async()` as
the Celery `task_id`, making it deterministic from DB state. Other
executors ignore it and overwrite with their own ID (e.g. ECS task
ARN) during event processing.
This also fixes the separate race in #55004 where `external_executor_id`
is lost when the task instance row is locked during event processing.
`process_executor_events` uses `skip_locked=True`, and
`get_event_buffer()` flushes the executor's in-memory buffer into a
local variable. If a TI is locked and skipped, its QUEUED event is
consumed from the buffer but never processed — the event and its
task ID are silently dropped. With the ID now written to the database
before the task is even sent to Celery, adoption no longer depends on
the event being processed.
Closes: #55004
Closes: #58570
Closes: #64971
---
.../src/airflow/executors/base_executor.py | 7 +-
.../src/airflow/executors/workloads/task.py | 1 +
.../src/airflow/jobs/scheduler_job_runner.py | 68 +++++++++++++---
.../src/airflow/utils/orm_event_handlers.py | 2 +
airflow-core/src/airflow/utils/sqlalchemy.py | 31 ++++++-
airflow-core/tests/unit/jobs/test_scheduler_job.py | 34 +++++++-
.../providers/celery/executors/celery_executor.py | 32 ++++----
.../celery/executors/celery_executor_utils.py | 11 ++-
.../unit/celery/executors/test_celery_executor.py | 95 ++++++++++++++++++++++
9 files changed, 252 insertions(+), 29 deletions(-)
diff --git a/airflow-core/src/airflow/executors/base_executor.py
b/airflow-core/src/airflow/executors/base_executor.py
index 36d8c86f5b7..71d43a76a35 100644
--- a/airflow-core/src/airflow/executors/base_executor.py
+++ b/airflow-core/src/airflow/executors/base_executor.py
@@ -24,7 +24,7 @@ from collections import defaultdict, deque
from collections.abc import Sequence
from dataclasses import dataclass, field
from functools import cached_property
-from typing import TYPE_CHECKING, Any, cast
+from typing import TYPE_CHECKING, Any, ClassVar, cast
import pendulum
@@ -173,6 +173,11 @@ class BaseExecutor(LoggingMixin):
is_local: bool = False
is_production: bool = True
+ # When True, the scheduler pre-assigns external_executor_id (a UUID) at
queuing time,
+ # committed atomically with the QUEUED state. The executor can then use
this ID to
+ # correlate the task with its external representation (e.g. Celery
task_id).
+ pre_assigns_external_executor_id: ClassVar[bool] = False
+
serve_logs: bool = False
job_id: None | int | str = None
diff --git a/airflow-core/src/airflow/executors/workloads/task.py
b/airflow-core/src/airflow/executors/workloads/task.py
index 3a95aab8310..d05affe4330 100644
--- a/airflow-core/src/airflow/executors/workloads/task.py
+++ b/airflow-core/src/airflow/executors/workloads/task.py
@@ -48,6 +48,7 @@ class TaskInstanceDTO(BaseModel):
queue: str
priority_weight: int
executor_config: dict | None = Field(default=None, exclude=True)
+ external_executor_id: str | None = Field(default=None, exclude=True)
parent_context_carrier: dict | None = None
context_carrier: dict | None = None
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index e9db7625c38..67115ba7578 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -33,7 +33,7 @@ from functools import lru_cache, partial
from itertools import groupby
from typing import TYPE_CHECKING, Any, cast
-from sqlalchemy import CTE, and_, delete, exists, func, inspect, or_, select,
text, tuple_, update
+from sqlalchemy import CTE, and_, case, delete, exists, func, inspect, or_,
select, text, tuple_, update
from sqlalchemy.exc import DBAPIError, OperationalError
from sqlalchemy.orm import joinedload, lazyload, load_only, make_transient,
selectinload
from sqlalchemy.sql import expression
@@ -98,6 +98,7 @@ from airflow.utils.sqlalchemy import (
get_dialect_name,
is_lock_not_available_error,
prohibit_commit,
+ random_db_uuid,
with_row_locks,
)
from airflow.utils.state import CallbackState, DagRunState, State,
TaskInstanceState
@@ -909,19 +910,66 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
filter_for_tis = TI.filter_for_tis(executable_tis)
if filter_for_tis is None:
return []
- session.execute(
+
+ queued_values: dict[str, Any] = {
+ "state": TaskInstanceState.QUEUED,
+ "queued_dttm": timezone.utcnow(),
+ "queued_by_job_id": self.job.id,
+ }
+
+ # Pre-assign external_executor_id atomically with the QUEUED state
so it
+ # survives a scheduler crash. Only done when an executor opts in
via
+ # pre_assigns_external_executor_id (e.g. CeleryExecutor uses it as
the
+ # Celery task_id passed to apply_async). In mixed-executor
deployments,
+ # a CASE expression limits the UUID to TIs targeting an opt-in
executor.
+ pre_assign_executors = {e for e in self.executors if
e.pre_assigns_external_executor_id}
+ if pre_assign_executors == set(self.executors):
+ # All executors opt in — unconditional UUID for every TI.
+ queued_values["external_executor_id"] = random_db_uuid()
+ elif pre_assign_executors:
+ # Mixed — only TIs routed to an opt-in executor get a UUID.
+ opt_in_names: set[str] = set()
+ default_opts_in = self.executor in pre_assign_executors
+ for exc in pre_assign_executors:
+ if exc.name:
+ if exc.name.alias:
+ opt_in_names.add(exc.name.alias)
+ opt_in_names.add(exc.name.module_path)
+ whens = []
+ if opt_in_names:
+ whens.append((TI.executor.in_(opt_in_names),
random_db_uuid()))
+ if default_opts_in:
+ whens.append((TI.executor.is_(None), random_db_uuid()))
+ if whens:
+ queued_values["external_executor_id"] = case(*whens,
else_=TI.external_executor_id)
+
+ queued_update = (
update(TI)
.where(filter_for_tis)
- .values(
- # TODO[ha]: should we use func.now()? How does that work
with DB timezone
- # on mysql when it's not UTC?
- state=TaskInstanceState.QUEUED,
- queued_dttm=timezone.utcnow(),
- queued_by_job_id=self.job.id,
- )
+ .values(**queued_values)
.execution_options(synchronize_session=False)
)
+ if pre_assign_executors:
+ # Read the DB-generated UUIDs back onto the in-memory objects
so the
+ # workload DTO carries them through to
send_workload_to_executor (the
+ # objects are about to be detached by make_transient). Use
RETURNING
+ # where supported (PostgreSQL); fall back to a SELECT for
MySQL and
+ # SQLite (RETURNING requires SQLite 3.35+ which isn't
guaranteed).
+ if get_dialect_name(session) == "postgresql":
+ result = session.execute(queued_update.returning(TI.id,
TI.external_executor_id))
+ id_map = {row[0]: row[1] for row in result}
+ else:
+ session.execute(queued_update)
+ id_rows = session.execute(
+ select(TI.id,
TI.external_executor_id).where(filter_for_tis)
+ ).all()
+ id_map = {row[0]: row[1] for row in id_rows}
+ for ti in executable_tis:
+ ti.external_executor_id = id_map.get(ti.id)
+ else:
+ session.execute(queued_update)
+
for ti in executable_tis:
ti.emit_state_change_metric(TaskInstanceState.QUEUED)
@@ -2766,7 +2814,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
.where(Job.state.is_distinct_from(JobState.RUNNING))
.join(TI.dag_run)
.where(DagRun.state == DagRunState.RUNNING)
- .options(load_only(TI.dag_id, TI.task_id, TI.run_id))
+ .options(load_only(TI.dag_id, TI.task_id, TI.run_id,
TI.external_executor_id))
)
# Lock these rows, so that another scheduler can't try and
adopt these too
diff --git a/airflow-core/src/airflow/utils/orm_event_handlers.py
b/airflow-core/src/airflow/utils/orm_event_handlers.py
index bbbd2a21d7e..da42433dbb0 100644
--- a/airflow-core/src/airflow/utils/orm_event_handlers.py
+++ b/airflow-core/src/airflow/utils/orm_event_handlers.py
@@ -21,6 +21,7 @@ import logging
import os
import time
import traceback
+from uuid import uuid4
from sqlalchemy import event, exc
from sqlalchemy.orm import Mapper
@@ -48,6 +49,7 @@ def setup_event_handlers(engine):
cursor.execute("PRAGMA foreign_keys=ON")
cursor.execute("PRAGMA journal_mode=WAL")
cursor.close()
+ dbapi_connection.create_function("uuid4", 0, lambda: str(uuid4()))
# this ensures coherence in mysql when storing datetimes (not required for
postgres)
if engine.dialect.name == "mysql":
diff --git a/airflow-core/src/airflow/utils/sqlalchemy.py
b/airflow-core/src/airflow/utils/sqlalchemy.py
index 266be08c3bb..a767c65fad9 100644
--- a/airflow-core/src/airflow/utils/sqlalchemy.py
+++ b/airflow-core/src/airflow/utils/sqlalchemy.py
@@ -24,9 +24,11 @@ import logging
from collections.abc import Generator
from typing import TYPE_CHECKING
-from sqlalchemy import TIMESTAMP, PickleType, event, nullsfirst
+from sqlalchemy import TIMESTAMP, PickleType, String, event, nullsfirst
from sqlalchemy.dialects import mysql
from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.ext.compiler import compiles
+from sqlalchemy.sql.functions import FunctionElement
from sqlalchemy.types import JSON, Text, TypeDecorator
from airflow._shared.timezones.timezone import make_naive, utc
@@ -56,6 +58,33 @@ def get_dialect_name(session: Session) -> str | None:
return getattr(bind.dialect, "name", None)
+class random_db_uuid(FunctionElement):
+ """
+ Cross-dialect random UUID generation for use in SQL expressions.
+
+ Compiles to ``gen_random_uuid()`` on PostgreSQL, ``UUID()`` on MySQL,
+ and ``uuid4()`` on SQLite (registered via :func:`setup_event_handlers`).
+ """
+
+ type = String()
+ inherit_cache = True
+
+
+@compiles(random_db_uuid, "postgresql")
+def _random_db_uuid_pg(element, compiler, **kw):
+ return "gen_random_uuid()"
+
+
+@compiles(random_db_uuid, "mysql")
+def _random_db_uuid_mysql(element, compiler, **kw):
+ return "UUID()"
+
+
+@compiles(random_db_uuid, "sqlite")
+def _random_db_uuid_sqlite(element, compiler, **kw):
+ return "uuid4()"
+
+
class UtcDateTime(TypeDecorator):
"""
Similar to :class:`~sqlalchemy.types.TIMESTAMP` with ``timezone=True``
option, with some differences.
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py
b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index 84694b47c69..061a87e3aa4 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -30,7 +30,7 @@ from pathlib import Path
from typing import TYPE_CHECKING
from unittest import mock
from unittest.mock import MagicMock, patch
-from uuid import uuid4
+from uuid import UUID, uuid4
import pendulum
import psutil
@@ -2421,6 +2421,38 @@ class TestSchedulerJob:
assert mock_queue_workload.called
session.rollback()
+ def
test_executable_task_instances_to_queued_sets_external_executor_id(self,
dag_maker, session):
+ """external_executor_id is written to the DB in the same UPDATE that
sets state=QUEUED."""
+ dag_id = "SchedulerJobTest.test_executable_sets_external_executor_id"
+ session = settings.Session()
+ with dag_maker(dag_id=dag_id, start_date=DEFAULT_DATE,
session=session):
+ EmptyOperator(task_id="dummy")
+
+ class PreAssigningExecutor(MockExecutor):
+ pre_assigns_external_executor_id = True
+
+ scheduler_job = Job()
+ self.job_runner = SchedulerJobRunner(job=scheduler_job,
executors=[PreAssigningExecutor()])
+
+ dr = dag_maker.create_dagrun()
+ ti = dr.get_task_instance("dummy", session)
+ ti.state = State.SCHEDULED
+ session.merge(ti)
+ session.flush()
+
+ returned_tis =
self.job_runner._executable_task_instances_to_queued(max_tis=32,
session=session)
+
+ assert len(returned_tis) == 1
+ # In-memory object (post make_transient) should carry the UUID
+ assert returned_tis[0].external_executor_id is not None
+ UUID(returned_tis[0].external_executor_id)
+
+ # DB row should also have it (the whole point — survives a crash)
+ db_value =
session.scalar(select(TaskInstance.external_executor_id).where(TaskInstance.id
== ti.id))
+ assert db_value == returned_tis[0].external_executor_id
+
+ session.rollback()
+
@pytest.mark.parametrize("state", [State.FAILED, State.SUCCESS])
def
test_enqueue_task_instances_sets_ti_state_to_None_if_dagrun_in_finish_state(self,
state, dag_maker):
"""This tests that task instances whose dagrun is in finished state
are not queued"""
diff --git
a/providers/celery/src/airflow/providers/celery/executors/celery_executor.py
b/providers/celery/src/airflow/providers/celery/executors/celery_executor.py
index d30bca1e570..16d634fbdef 100644
--- a/providers/celery/src/airflow/providers/celery/executors/celery_executor.py
+++ b/providers/celery/src/airflow/providers/celery/executors/celery_executor.py
@@ -32,7 +32,7 @@ import time
from collections import Counter
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import cpu_count
-from typing import TYPE_CHECKING, Any, TypeAlias, cast
+from typing import TYPE_CHECKING, Any, ClassVar, TypeAlias, cast
from celery import states as celery_states
from deprecated import deprecated
@@ -108,6 +108,7 @@ class CeleryExecutor(BaseExecutor):
supports_ad_hoc_ti_run: bool = True
supports_callbacks: bool = True
sentry_integration: str =
"sentry_sdk.integrations.celery.CeleryIntegration"
+ pre_assigns_external_executor_id: ClassVar[bool] = True
# TODO: Remove this flag once providers depend on Airflow 3.2.
supports_sentry: bool = True
@@ -312,21 +313,22 @@ class CeleryExecutor(BaseExecutor):
pass
def try_adopt_task_instances(self, tis: Sequence[TaskInstance]) ->
Sequence[TaskInstance]:
- # See which of the TIs are still alive (or have finished even!)
+ # The scheduler pre-assigns external_executor_id at queuing time
(committed to DB
+ # atomically with the QUEUED state), and the same ID is used as the
Celery task_id
+ # in apply_async(). This means external_executor_id is always set for
tasks queued
+ # by current code, and always matches the Celery task.
#
- # 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.
+ # A narrow window remains: the scheduler could die after committing
the QUEUED
+ # state (with the pre-assigned ID) but before executor.heartbeat()
calls
+ # apply_async(). In that case the ID is in the DB but the Celery task
was never
+ # submitted. AsyncResult returns PENDING and the task is adopted and
monitored
+ # until task_queued_timeout fires. This is an acceptable trade-off:
the window is
+ # small (within a single scheduler loop iteration) and the task is
eventually
+ # recovered, whereas the pre-fix behaviour caused duplicate execution.
#
- # 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.)
+ # Tasks with external_executor_id=None were queued by older scheduler
code.
+ # We cannot adopt these — there is no Celery task_id to look up — so
they go to
+ # not_adopted_tis and get reset by the caller.
from celery.result import AsyncResult
celery_tasks = {}
@@ -352,8 +354,8 @@ class CeleryExecutor(BaseExecutor):
result, ti = celery_tasks[celery_task_id]
result.backend = cached_celery_backend
if isinstance(result.result, BaseException):
- e = result.result
# Log the exception we got from the remote end
+ e = result.result
self.log.warning("Task %s failed with error", ti.key,
exc_info=e)
# Set the correct elements of the state dicts, then update this
diff --git
a/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
b/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
index 5ff62866e8c..f16b7c226fa 100644
---
a/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
+++
b/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
@@ -408,11 +408,20 @@ def send_workload_to_executor(
# Create the Celery app with the correct configuration.
celery_app = create_celery_app(_conf)
+ celery_task_id = None
if AIRFLOW_V_3_0_PLUS:
# Get the task from the app.
celery_task = celery_app.tasks["execute_workload"]
if TYPE_CHECKING:
assert isinstance(args, workloads.BaseWorkload)
+ # Extract the pre-assigned Celery task ID before serializing the
workload.
+ # This ID was committed to the DB at queuing time (as
external_executor_id) and is
+ # excluded from model_dump_json(), so workers never see it. Passing it
to apply_async()
+ # makes the Celery task ID deterministic from DB state, closing the
race window where a
+ # scheduler crash between apply_async() and event processing left
external_executor_id
+ # unset and the task unadoptable.
+ if executor_id := getattr(getattr(args, "ti", None),
"external_executor_id", None):
+ celery_task_id = executor_id
args = (args.model_dump_json(),)
else:
# Get the task from the app.
@@ -430,7 +439,7 @@ def send_workload_to_executor(
try:
with timeout(seconds=OPERATION_TIMEOUT):
- result = celery_task.apply_async(args=args, queue=queue)
+ result = celery_task.apply_async(args=args, queue=queue,
task_id=celery_task_id)
except (Exception, AirflowTaskTimeout) as e:
exception_traceback = f"Celery Task ID:
{key}\n{traceback.format_exc()}"
result = ExceptionWithTraceback(e, exception_traceback)
diff --git
a/providers/celery/tests/unit/celery/executors/test_celery_executor.py
b/providers/celery/tests/unit/celery/executors/test_celery_executor.py
index 438c9487192..7b1ec702fc4 100644
--- a/providers/celery/tests/unit/celery/executors/test_celery_executor.py
+++ b/providers/celery/tests/unit/celery/executors/test_celery_executor.py
@@ -286,6 +286,39 @@ class TestCeleryExecutor:
assert executor.workloads == {key_1: AsyncResult("231"), key_2:
AsyncResult("232")}
assert not_adopted_tis == []
+ @pytest.mark.backend("mysql", "postgres")
+ @time_machine.travel("2020-01-01", tick=False)
+ def test_try_adopt_with_and_without_external_executor_id(
+ self, clean_dags_dagruns_and_dagbundles, testing_dag_bundle
+ ):
+ """TIs with an ID are adopted; TIs without are returned for reset."""
+ start_date = timezone.utcnow() - timedelta(days=2)
+
+ with DAG("test_try_adopt_mixed", schedule=None) as dag:
+ task_1 = BaseOperator(task_id="task_1", start_date=start_date)
+ task_2 = BaseOperator(task_id="task_2", start_date=start_date)
+
+ if AIRFLOW_V_3_0_PLUS:
+ sync_dag_to_db(dag)
+ dag_version = DagVersion.get_latest_version(dag.dag_id)
+ ti_with_id = create_task_instance(task=task_1, run_id=None,
dag_version_id=dag_version.id)
+ ti_without_id = create_task_instance(task=task_2, run_id=None,
dag_version_id=dag_version.id)
+ else:
+ ti_with_id = TaskInstance(task=task_1, run_id=None)
+ ti_without_id = TaskInstance(task=task_2, run_id=None)
+ ti_with_id.external_executor_id = "231"
+ ti_with_id.state = State.QUEUED
+ ti_without_id.external_executor_id = None
+ ti_without_id.state = State.QUEUED
+
+ executor = celery_executor.CeleryExecutor()
+ not_adopted_tis = executor.try_adopt_task_instances([ti_with_id,
ti_without_id])
+
+ key_1 = TaskInstanceKey(dag.dag_id, task_1.task_id, None, 0)
+ assert key_1 in executor.running
+ assert executor.workloads == {key_1: AsyncResult("231")}
+ assert not_adopted_tis == [ti_without_id]
+
@pytest.fixture
def mock_celery_revoke(self):
with _prepare_app() as app:
@@ -439,6 +472,68 @@ def test_send_workloads_to_celery_hang(register_signals):
assert results == [(None, None, 1) for _ in workload_tuples_to_send]
+def _has_external_executor_id_field() -> bool:
+ try:
+ from airflow.executors.workloads.task import TaskInstanceDTO
+
+ return "external_executor_id" in TaskInstanceDTO.model_fields
+ except (ImportError, AttributeError):
+ return False
+
+
[email protected](
+ not _has_external_executor_id_field(),
+ reason="TaskInstanceDTO.external_executor_id not available in this
airflow-core version",
+)
+def test_send_workload_uses_external_executor_id_as_celery_task_id():
+ """Pre-assigned external_executor_id is passed as task_id to
apply_async()."""
+ from airflow.executors.workloads.task import TaskInstanceDTO
+
+ pre_assigned_id = "pre-assigned-uuid-for-celery"
+ ti_dto = TaskInstanceDTO(
+ id="00000000-0000-0000-0000-000000000001",
+ dag_version_id="00000000-0000-0000-0000-000000000002",
+ task_id="test_task",
+ dag_id="test_dag",
+ run_id="test_run",
+ try_number=1,
+ map_index=-1,
+ pool_slots=1,
+ queue="default",
+ priority_weight=1,
+ external_executor_id=pre_assigned_id,
+ )
+ key = TaskInstanceKey(
+ dag_id="test_dag", task_id="test_task", run_id="test_run",
map_index=-1, try_number=1
+ )
+
+ mock_result = mock.Mock(task_id=pre_assigned_id)
+ mock_celery_task = mock.Mock()
+ mock_celery_task.apply_async.return_value = mock_result
+
+ mock_app = mock.Mock()
+ mock_app.tasks = {"execute_workload": mock_celery_task}
+
+ from airflow.executors.workloads import ExecuteTask
+
+ workload = mock.Mock(spec=ExecuteTask)
+ workload.ti = ti_dto
+ workload.model_dump_json.return_value = "{}"
+
+ with mock.patch(
+
"airflow.providers.celery.executors.celery_executor_utils.create_celery_app",
+ return_value=mock_app,
+ ):
+ result_key, _, result =
celery_executor_utils.send_workload_to_executor(
+ (key, workload, "default", None)
+ )
+
+ mock_celery_task.apply_async.assert_called_once_with(
+ args=("{}",), queue="default", task_id=pre_assigned_id
+ )
+ assert result.task_id == pre_assigned_id
+
+
@conf_vars({("celery", "result_backend"):
"rediss://test_user:test_password@localhost:6379/0"})
def test_celery_executor_with_no_recommended_result_backend(caplog):
import importlib
