This is an automated email from the ASF dual-hosted git repository.
jscheffl pushed a commit to branch v3-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-1-test by this push:
new 5c629259b96 [v3-1-test] Add no-op _process_workloads to EdgeExecutor
to improve readability (#64236) (#64508)
5c629259b96 is described below
commit 5c629259b96a50f314b3882df899c70ff76a8661
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Mon Mar 30 23:19:53 2026 +0200
[v3-1-test] Add no-op _process_workloads to EdgeExecutor to improve
readability (#64236) (#64508)
* Add no-op _process_workloads to EdgeExecutor to improve EdgeExecutor
readability
* add docs
(cherry picked from commit 476dabe4d71bc089993e092c3cb1b01e75681377)
Co-authored-by: Jeongwoo Do <[email protected]>
---
airflow-core/docs/core-concepts/executor/index.rst | 1 +
.../edge3/src/airflow/providers/edge3/executors/edge_executor.py | 9 +++++++++
2 files changed, 10 insertions(+)
diff --git a/airflow-core/docs/core-concepts/executor/index.rst
b/airflow-core/docs/core-concepts/executor/index.rst
index 469bbf0847e..5dfce3f2c7b 100644
--- a/airflow-core/docs/core-concepts/executor/index.rst
+++ b/airflow-core/docs/core-concepts/executor/index.rst
@@ -288,6 +288,7 @@ The following methods must be overridden at minimum to have
your executor suppor
* ``sync``: Sync will get called periodically during executor heartbeats.
Implement this method to update the state of the tasks which the executor knows
about. Optionally, attempting to execute queued tasks that have been received
from the scheduler.
* ``execute_async``: Executes a *workload* asynchronously. This method is
called (after a few layers) during executor heartbeat which is run periodically
by the scheduler. In practice, this method often just enqueues tasks into an
internal or external queue of tasks to be run (e.g. ``KubernetesExecutor``).
But can also execute the tasks directly as well (e.g. ``LocalExecutor``). This
will depend on the executor.
+* ``_process_workloads``: Processes a list of workloads that have been queued
via ``queue_workload``. This method is called during executor heartbeat and
defines how the executor handles the execution of workloads (e.g., queuing them
to workers, submitting to external systems, etc.).
Optional Interface Methods to Implement
diff --git
a/providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py
b/providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py
index 6b15cd342cc..e288b9958b1 100644
--- a/providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py
+++ b/providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py
@@ -222,6 +222,15 @@ class EdgeExecutor(BaseExecutor):
)
)
+ def _process_workloads(self, workloads: Sequence[workloads.All]) -> None:
+ """
+ No-op: EdgeExecutor does not use the BaseExecutor workload pipeline.
+
+ EdgeExecutor handles task queuing directly in queue_workload() by
writing
+ to the EdgeJobModel database table, bypassing BaseExecutor's
queued_tasks.
+ Therefore, trigger_tasks() never accumulates workloads to pass here.
+ """
+
def _check_worker_liveness(self, session: Session) -> bool:
"""Reset worker state if heartbeat timed out."""
changed = False
