This is an automated email from the ASF dual-hosted git repository.
shahar1 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 2cc7bb0c702 Fix scheduler crash loop from large Edge deadline
callbacks (#74445)
2cc7bb0c702 is described below
commit 2cc7bb0c70273f8265811e93becde8144bc7b914
Author: KidAmnesiac1 <[email protected]>
AuthorDate: Sat Nov 7 01:13:06 2026 +0800
Fix scheduler crash loop from large Edge deadline callbacks (#74445)
A missed deadline with a SyncCallback is queued to the EdgeExecutor as
JSON in edge_job.command, which is limited to 2048 characters. The
callback carries a snapshot of the Dag run (including its conf), so it
can easily go over the limit; ours was about 2.3 KB. The database
rejects the insert, the scheduler loop fails and rolls back, and every
restart picks up the same deadline and crashes again, stopping
scheduling for the whole deployment.
---
providers/edge3/docs/migrations-ref.rst | 4 +-
.../versions/0008_5_2_0_widen_edge_job_command.py | 55 ++++++++++++++++++++++
.../edge3/src/airflow/providers/edge3/models/db.py | 1 +
.../src/airflow/providers/edge3/models/edge_job.py | 3 +-
.../unit/edge3/executors/test_edge_executor.py | 13 +++++
5 files changed, 74 insertions(+), 2 deletions(-)
diff --git a/providers/edge3/docs/migrations-ref.rst
b/providers/edge3/docs/migrations-ref.rst
index e4ade8ae266..f20bb22ad60 100644
--- a/providers/edge3/docs/migrations-ref.rst
+++ b/providers/edge3/docs/migrations-ref.rst
@@ -34,7 +34,9 @@ Here's the list of all the Database Migrations that are
executed via when you ru
+-------------------------+------------------+-----------------+----------------------------------------------------------+
| Revision ID | Revises ID | Edge3 Version | Description
|
+=========================+==================+=================+==========================================================+
-| ``d5a2f8b41c07`` (head) | ``f2a4b6c8d0e1`` | ``5.1.0`` | Add
priority_weight column to edge_job table. |
+| ``cd96a0a0e458`` (head) | ``d5a2f8b41c07`` | ``5.2.0`` | Change
edge_job.command to unbounded text. |
++-------------------------+------------------+-----------------+----------------------------------------------------------+
+| ``d5a2f8b41c07`` | ``f2a4b6c8d0e1`` | ``5.1.0`` | Add
priority_weight column to edge_job table. |
+-------------------------+------------------+-----------------+----------------------------------------------------------+
| ``f2a4b6c8d0e1`` | ``c6b3c3d093fd`` | ``5.0.0`` | Add task
instance identity to Edge jobs. |
+-------------------------+------------------+-----------------+----------------------------------------------------------+
diff --git
a/providers/edge3/src/airflow/providers/edge3/migrations/versions/0008_5_2_0_widen_edge_job_command.py
b/providers/edge3/src/airflow/providers/edge3/migrations/versions/0008_5_2_0_widen_edge_job_command.py
new file mode 100644
index 00000000000..52ae26af5b5
--- /dev/null
+++
b/providers/edge3/src/airflow/providers/edge3/migrations/versions/0008_5_2_0_widen_edge_job_command.py
@@ -0,0 +1,55 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Change edge_job.command to unbounded text.
+
+Revision ID: cd96a0a0e458
+Revises: d5a2f8b41c07
+Create Date: 2026-10-08 00:00:00.000000
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects.mysql import MEDIUMTEXT
+
+revision = "cd96a0a0e458"
+down_revision = "d5a2f8b41c07"
+branch_labels = None
+depends_on = None
+edge3_version = "5.2.0"
+
+
+def upgrade() -> None:
+ with op.batch_alter_table("edge_job") as batch_op:
+ batch_op.alter_column(
+ "command",
+ existing_type=sa.String(length=2048),
+ type_=sa.Text().with_variant(MEDIUMTEXT(), "mysql"),
+ existing_nullable=False,
+ )
+
+
+def downgrade() -> None:
+ with op.batch_alter_table("edge_job") as batch_op:
+ batch_op.alter_column(
+ "command",
+ existing_type=sa.Text().with_variant(MEDIUMTEXT(), "mysql"),
+ type_=sa.String(length=2048),
+ existing_nullable=False,
+ )
diff --git a/providers/edge3/src/airflow/providers/edge3/models/db.py
b/providers/edge3/src/airflow/providers/edge3/models/db.py
index 315b8397d81..efa5d5d6844 100644
--- a/providers/edge3/src/airflow/providers/edge3/models/db.py
+++ b/providers/edge3/src/airflow/providers/edge3/models/db.py
@@ -49,6 +49,7 @@ _REVISION_HEADS_MAP: dict[str, str] = {
"3.5.0": "c6b3c3d093fd",
"5.0.0": "f2a4b6c8d0e1",
"5.1.0": "d5a2f8b41c07",
+ "5.2.0": "cd96a0a0e458",
}
diff --git a/providers/edge3/src/airflow/providers/edge3/models/edge_job.py
b/providers/edge3/src/airflow/providers/edge3/models/edge_job.py
index 41c8c2f5c17..fd7b977beb3 100644
--- a/providers/edge3/src/airflow/providers/edge3/models/edge_job.py
+++ b/providers/edge3/src/airflow/providers/edge3/models/edge_job.py
@@ -23,6 +23,7 @@ from sqlalchemy import (
Index,
Integer,
String,
+ Text,
text,
)
from sqlalchemy.dialects import mysql
@@ -85,7 +86,7 @@ class EdgeJobModel(Base, LoggingMixin):
queue: Mapped[str] = mapped_column(String(256))
concurrency_slots: Mapped[int] = mapped_column(Integer)
priority_weight: Mapped[int] = mapped_column(Integer, nullable=False,
server_default=text("1"))
- command: Mapped[str] = mapped_column(String(2048))
+ command: Mapped[str] =
mapped_column(Text().with_variant(mysql.MEDIUMTEXT(), "mysql"))
queued_dttm: Mapped[datetime | None] = mapped_column(UtcDateTime)
edge_worker: Mapped[str | None] = mapped_column(String(64))
last_update: Mapped[datetime | None] = mapped_column(UtcDateTime)
diff --git a/providers/edge3/tests/unit/edge3/executors/test_edge_executor.py
b/providers/edge3/tests/unit/edge3/executors/test_edge_executor.py
index 023b248bd90..ecb6fbafa3e 100644
--- a/providers/edge3/tests/unit/edge3/executors/test_edge_executor.py
+++ b/providers/edge3/tests/unit/edge3/executors/test_edge_executor.py
@@ -1105,6 +1105,19 @@ class TestQueueWorkload(_WorkloadFactory):
assert job.state == TaskInstanceState.QUEUED
assert '"type":"ExecuteCallback"' in job.command or '"type":
"ExecuteCallback"' in job.command
+ def test_queue_workload_stores_command_longer_than_2048_chars(self):
+ executor = EdgeExecutor()
+ workload = self._make_execute_callback()
+ workload.callback.data["kwargs"]["context"] = {"dag_run": {"conf":
{"payload": "x" * 4096}}}
+
+ with create_session() as session:
+ executor.queue_workload(workload, session=session)
+
+ with create_session() as session:
+ job = session.scalar(select(EdgeJobModel))
+ assert job is not None
+ assert job.command == workload.model_dump_json()
+
def test_queue_workload_execute_callback_existing_job(self):
executor = EdgeExecutor()
workload = self._make_execute_callback()
