You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Prasanth345 (via GitHub)" <gi...@apache.org> on 2025/07/30 12:54:25 UTC
[PR] Fix: Prevent duplicate edge_job insertions for deferrable tasks in EdgeExecutor (#53610) [airflow]
Prasanth345 opened a new pull request, #53927:
URL: https://github.com/apache/airflow/pull/53927
## Overview
This PR fixes a crash in the Apache Airflow `EdgeExecutor` caused by duplicate entries in the `edge_job` table when handling deferrable tasks like `ExternalTaskSensor(mode="reschedule")`.
---
## Root Cause
When a deferrable task times out and retries, both `execute_async()` and `queue_workload()` attempt to insert the same job into the `edge_job` table again, violating uniqueness constraints and causing the scheduler to crash.
---
## Solution
Instead of blindly inserting, both methods now:
- Check if a matching job exists (`dag_id`, `task_id`, `run_id`, `map_index`, `try_number`)
- If it exists → **Update** the job to reflect the latest state
- If not → Insert a new record
This preserves job queue integrity and ensures tasks are picked up correctly by the edge worker.
---
## Changes Made
- Patched `execute_async()` and `queue_workload()` in:
`airflow/providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py`
---
## Related Issue
Closes [#53610](https://github.com/apache/airflow/issues/53610)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Prevent duplicate edge_job insertions for deferrable tasks in EdgeExecutor (#53610) [airflow]
Posted by "Prasanth345 (via GitHub)" <gi...@apache.org>.
Prasanth345 commented on code in PR #53927:
URL: https://github.com/apache/airflow/pull/53927#discussion_r2249390199
##########
providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py:
##########
@@ -139,21 +139,46 @@ def execute_async(
task_instance = self.edge_queued_tasks[key][3] # type: ignore[index]
del self.edge_queued_tasks[key]
- self.validate_airflow_tasks_run_command(command) # type: ignore[attr-defined]
- session.add(
- EdgeJobModel(
+ # Run validation only if supported (available in Airflow 2.12+)
+ # This makes it compatible with older versions like 2.10 and 2.11
+ if hasattr(self, "validate_airflow_tasks_run_command"):
+ self.validate_airflow_tasks_run_command(command) # type: ignore[attr-defined]
+
Review Comment:
Thanks for the clarification!
I’ve updated the test case to use a valid Airflow task command (starting with ["airflow", "tasks", "run"]) and removed the conditional check around validate_airflow_tasks_run_command. The validation now behaves consistently as intended.
Let me know if there’s anything else you'd like me to address.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Prevent duplicate edge_job insertions for deferrable tasks in EdgeExecutor (#53610) [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #53927:
URL: https://github.com/apache/airflow/pull/53927#discussion_r2248794540
##########
providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py:
##########
@@ -139,21 +139,46 @@ def execute_async(
task_instance = self.edge_queued_tasks[key][3] # type: ignore[index]
del self.edge_queued_tasks[key]
- self.validate_airflow_tasks_run_command(command) # type: ignore[attr-defined]
- session.add(
- EdgeJobModel(
+ # Run validation only if supported (available in Airflow 2.12+)
+ # This makes it compatible with older versions like 2.10 and 2.11
+ if hasattr(self, "validate_airflow_tasks_run_command"):
+ self.validate_airflow_tasks_run_command(command) # type: ignore[attr-defined]
+
Review Comment:
Sorry, this is not good. The failure of tests in https://github.com/apache/airflow/actions/runs/16641791391/job/47152038655?pr=53927 was caused by an error in the test as you pass `executor.execute_async(key=key, command=["new", "command"])` in https://github.com/apache/airflow/pull/53927/files#diff-a195d35658cf228806ec7f8f959fc1a6dcdb67b8f2d1de6183289635a91ee72aR371 which is not a valid airflow task command. As the error in pytests points to `ValueError: The command must start with ["airflow", "tasks", "run"].` you need to pass a command in the pytest which is compatible to airflow task executions.
You should not change the validation behaviro for tasks in here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Prevent duplicate edge_job insertions for deferrable tasks in EdgeExecutor (#53610) [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on PR #53927:
URL: https://github.com/apache/airflow/pull/53927#issuecomment-3141271189
> @jscheffl Added test coverage under test_edge_executor.py to cover the updated behavior and prevent future regression.
Thanks!
Just a small glitch of asserts in tests in Airflow 2.10/2.11 - can you fix this? The command is validated and the mocked test fails in your pytest. Then I think it is good to be merged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Prevent duplicate edge_job insertions for deferrable tasks in EdgeExecutor (#53610) [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl merged PR #53927:
URL: https://github.com/apache/airflow/pull/53927
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Prevent duplicate edge_job insertions for deferrable tasks in EdgeExecutor (#53610) [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #53927:
URL: https://github.com/apache/airflow/pull/53927#discussion_r2250097350
##########
providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py:
##########
@@ -140,20 +140,42 @@ def execute_async(
del self.edge_queued_tasks[key]
self.validate_airflow_tasks_run_command(command) # type: ignore[attr-defined]
- session.add(
- EdgeJobModel(
+
+ # Check if job already exists with same dag_id, task_id, run_id, map_index, try_number
+ existing_job = (
+ session.query(EdgeJobModel)
+ .filter_by(
dag_id=key.dag_id,
task_id=key.task_id,
run_id=key.run_id,
map_index=key.map_index,
try_number=key.try_number,
- state=TaskInstanceState.QUEUED,
- queue=queue or DEFAULT_QUEUE,
- concurrency_slots=task_instance.pool_slots,
- command=str(command),
)
+ .first()
)
+ if existing_job:
+ # self.log.info(f"EdgeExecutor: Job already exists for {key}, updating it.")
+ existing_job.state = TaskInstanceState.QUEUED
+ existing_job.queue = queue or DEFAULT_QUEUE
+ existing_job.concurrency_slots = task_instance.pool_slots
+ existing_job.command = str(command)
+ else:
+ session.add(
+ EdgeJobModel(
+ dag_id=key.dag_id,
+ task_id=key.task_id,
+ run_id=key.run_id,
+ map_index=key.map_index,
+ try_number=key.try_number,
+ state=TaskInstanceState.QUEUED,
+ queue=queue or DEFAULT_QUEUE,
+ concurrency_slots=task_instance.pool_slots,
+ command=str(command),
+ )
+ )
+ # self.log.info(f"EdgeExecutor: Inserted new job for {key}")
Review Comment:
And not to forget, for performance you should write it with lazy string formatting as `self.log.debug("EdgeExecutor: Inserted new job for %s", key)`
##########
providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py:
##########
@@ -140,20 +140,42 @@ def execute_async(
del self.edge_queued_tasks[key]
self.validate_airflow_tasks_run_command(command) # type: ignore[attr-defined]
- session.add(
- EdgeJobModel(
+
+ # Check if job already exists with same dag_id, task_id, run_id, map_index, try_number
+ existing_job = (
+ session.query(EdgeJobModel)
+ .filter_by(
dag_id=key.dag_id,
task_id=key.task_id,
run_id=key.run_id,
map_index=key.map_index,
try_number=key.try_number,
- state=TaskInstanceState.QUEUED,
- queue=queue or DEFAULT_QUEUE,
- concurrency_slots=task_instance.pool_slots,
- command=str(command),
)
+ .first()
)
+ if existing_job:
+ # self.log.info(f"EdgeExecutor: Job already exists for {key}, updating it.")
+ existing_job.state = TaskInstanceState.QUEUED
+ existing_job.queue = queue or DEFAULT_QUEUE
+ existing_job.concurrency_slots = task_instance.pool_slots
+ existing_job.command = str(command)
+ else:
+ session.add(
+ EdgeJobModel(
+ dag_id=key.dag_id,
+ task_id=key.task_id,
+ run_id=key.run_id,
+ map_index=key.map_index,
+ try_number=key.try_number,
+ state=TaskInstanceState.QUEUED,
+ queue=queue or DEFAULT_QUEUE,
+ concurrency_slots=task_instance.pool_slots,
+ command=str(command),
+ )
+ )
+ # self.log.info(f"EdgeExecutor: Inserted new job for {key}")
Review Comment:
Can you remove the commended-out debug (or keep then and make them log.debug() instead)?
##########
providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py:
##########
@@ -168,20 +190,42 @@ def queue_workload(
task_instance = workload.ti
key = task_instance.key
- session.add(
- EdgeJobModel(
+
+ # Check if job already exists with same dag_id, task_id, run_id, map_index, try_number
+ existing_job = (
+ session.query(EdgeJobModel)
+ .filter_by(
dag_id=key.dag_id,
task_id=key.task_id,
run_id=key.run_id,
map_index=key.map_index,
try_number=key.try_number,
- state=TaskInstanceState.QUEUED,
- queue=task_instance.queue,
- concurrency_slots=task_instance.pool_slots,
- command=workload.model_dump_json(),
)
+ .first()
)
+ if existing_job:
+ # self.log.info(f"EdgeExecutor: Workload already exists for {key}, updating it.")
+ existing_job.state = TaskInstanceState.QUEUED
+ existing_job.queue = task_instance.queue
+ existing_job.concurrency_slots = task_instance.pool_slots
+ existing_job.command = workload.model_dump_json()
+ else:
+ session.add(
+ EdgeJobModel(
+ dag_id=key.dag_id,
+ task_id=key.task_id,
+ run_id=key.run_id,
+ map_index=key.map_index,
+ try_number=key.try_number,
+ state=TaskInstanceState.QUEUED,
+ queue=task_instance.queue,
+ concurrency_slots=task_instance.pool_slots,
+ command=workload.model_dump_json(),
+ )
+ )
+ # self.log.info(f"EdgeExecutor: Inserted new workload for {key}")
Review Comment:
Same here, log.debug() would be good.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Prevent duplicate edge_job insertions for deferrable tasks in EdgeExecutor (#53610) [airflow]
Posted by "Prasanth345 (via GitHub)" <gi...@apache.org>.
Prasanth345 commented on PR #53927:
URL: https://github.com/apache/airflow/pull/53927#issuecomment-3144007146
@jscheffl Thanks for the feedback!
I’ve updated the code to conditionally call validate_airflow_tasks_run_command using a hasattr check. This ensures compatibility with Airflow 2.10 and 2.11, where the method may not be present.
All tests are now passing across supported versions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Prevent duplicate edge_job insertions for deferrable tasks in EdgeExecutor (#53610) [airflow]
Posted by "Prasanth345 (via GitHub)" <gi...@apache.org>.
Prasanth345 commented on code in PR #53927:
URL: https://github.com/apache/airflow/pull/53927#discussion_r2249390199
##########
providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py:
##########
@@ -139,21 +139,46 @@ def execute_async(
task_instance = self.edge_queued_tasks[key][3] # type: ignore[index]
del self.edge_queued_tasks[key]
- self.validate_airflow_tasks_run_command(command) # type: ignore[attr-defined]
- session.add(
- EdgeJobModel(
+ # Run validation only if supported (available in Airflow 2.12+)
+ # This makes it compatible with older versions like 2.10 and 2.11
+ if hasattr(self, "validate_airflow_tasks_run_command"):
+ self.validate_airflow_tasks_run_command(command) # type: ignore[attr-defined]
+
Review Comment:
@jscheffl Thanks for the clarification!
I’ve updated the test case to use a valid Airflow task command (starting with ["airflow", "tasks", "run"]) and removed the conditional check around validate_airflow_tasks_run_command. The validation now behaves consistently as intended.
Let me know if there’s anything else you'd like me to address.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Prevent duplicate edge_job insertions for deferrable tasks in EdgeExecutor (#53610) [airflow]
Posted by "Prasanth345 (via GitHub)" <gi...@apache.org>.
Prasanth345 commented on PR #53927:
URL: https://github.com/apache/airflow/pull/53927#issuecomment-3138737889
@jscheffl Added test coverage under test_edge_executor.py to cover the updated behavior and prevent future regression.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Prevent duplicate edge_job insertions for deferrable tasks in EdgeExecutor (#53610) [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #53927:
URL: https://github.com/apache/airflow/pull/53927#issuecomment-3136210949
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:
- Pay attention to the quality of your code (ruff, mypy and type annotations). Our [pre-commits]( https://github.com/apache/airflow/blob/main/contributing-docs/08_static_code_checks.rst#prerequisites-for-pre-commit-hooks) will help you with that.
- In case of a new feature add useful documentation (in docstrings or in `docs/` directory). Adding a new operator? Check this short [guide](https://github.com/apache/airflow/blob/main/airflow-core/docs/howto/custom-operator.rst) Consider adding an example DAG that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/main/dev/breeze/doc/README.rst) for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
- Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
- Please follow [ASF Code of Conduct](https://www.apache.org/foundation/policies/conduct) for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
- Be sure to read the [Airflow Coding style]( https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#coding-style-and-best-practices).
- Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Prevent duplicate edge_job insertions for deferrable tasks in EdgeExecutor (#53610) [airflow]
Posted by "Prasanth345 (via GitHub)" <gi...@apache.org>.
Prasanth345 commented on code in PR #53927:
URL: https://github.com/apache/airflow/pull/53927#discussion_r2250644038
##########
providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py:
##########
@@ -140,20 +140,42 @@ def execute_async(
del self.edge_queued_tasks[key]
self.validate_airflow_tasks_run_command(command) # type: ignore[attr-defined]
- session.add(
- EdgeJobModel(
+
+ # Check if job already exists with same dag_id, task_id, run_id, map_index, try_number
+ existing_job = (
+ session.query(EdgeJobModel)
+ .filter_by(
dag_id=key.dag_id,
task_id=key.task_id,
run_id=key.run_id,
map_index=key.map_index,
try_number=key.try_number,
- state=TaskInstanceState.QUEUED,
- queue=queue or DEFAULT_QUEUE,
- concurrency_slots=task_instance.pool_slots,
- command=str(command),
)
+ .first()
)
+ if existing_job:
+ # self.log.info(f"EdgeExecutor: Job already exists for {key}, updating it.")
+ existing_job.state = TaskInstanceState.QUEUED
+ existing_job.queue = queue or DEFAULT_QUEUE
+ existing_job.concurrency_slots = task_instance.pool_slots
+ existing_job.command = str(command)
+ else:
+ session.add(
+ EdgeJobModel(
+ dag_id=key.dag_id,
+ task_id=key.task_id,
+ run_id=key.run_id,
+ map_index=key.map_index,
+ try_number=key.try_number,
+ state=TaskInstanceState.QUEUED,
+ queue=queue or DEFAULT_QUEUE,
+ concurrency_slots=task_instance.pool_slots,
+ command=str(command),
+ )
+ )
+ # self.log.info(f"EdgeExecutor: Inserted new job for {key}")
Review Comment:
I’ve removed the commented-out debug logs to keep the code clean
##########
providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py:
##########
@@ -168,20 +190,42 @@ def queue_workload(
task_instance = workload.ti
key = task_instance.key
- session.add(
- EdgeJobModel(
+
+ # Check if job already exists with same dag_id, task_id, run_id, map_index, try_number
+ existing_job = (
+ session.query(EdgeJobModel)
+ .filter_by(
dag_id=key.dag_id,
task_id=key.task_id,
run_id=key.run_id,
map_index=key.map_index,
try_number=key.try_number,
- state=TaskInstanceState.QUEUED,
- queue=task_instance.queue,
- concurrency_slots=task_instance.pool_slots,
- command=workload.model_dump_json(),
)
+ .first()
)
+ if existing_job:
+ # self.log.info(f"EdgeExecutor: Workload already exists for {key}, updating it.")
+ existing_job.state = TaskInstanceState.QUEUED
+ existing_job.queue = task_instance.queue
+ existing_job.concurrency_slots = task_instance.pool_slots
+ existing_job.command = workload.model_dump_json()
+ else:
+ session.add(
+ EdgeJobModel(
+ dag_id=key.dag_id,
+ task_id=key.task_id,
+ run_id=key.run_id,
+ map_index=key.map_index,
+ try_number=key.try_number,
+ state=TaskInstanceState.QUEUED,
+ queue=task_instance.queue,
+ concurrency_slots=task_instance.pool_slots,
+ command=workload.model_dump_json(),
+ )
+ )
+ # self.log.info(f"EdgeExecutor: Inserted new workload for {key}")
Review Comment:
I’ve removed the commented-out debug logs to keep the code clean
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Prevent duplicate edge_job insertions for deferrable tasks in EdgeExecutor (#53610) [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #53927:
URL: https://github.com/apache/airflow/pull/53927#issuecomment-3152387069
Awesome work, congrats on your first merged pull request! You are invited to check our [Issue Tracker](https://github.com/apache/airflow/issues) for additional contributions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org