Fix: Prevent duplicate edge_job insertions for deferrable tasks in EdgeExecutor (#53610)

[Prasanth345]

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

[boring-cyborg]

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 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 Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment 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 for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• 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

[jscheffl]

VEry good! Thanks for the fix! I can confirm this resolved the problem!

To be merged, can you please also extend the existing use cases in providers/edge3/tests/unit/edge3/executors/test_edge_executor.py to prevent a future regression?

[Prasanth345]

@jscheffl Added test coverage under test_edge_executor.py to cover the updated behavior and prevent future regression.

[jscheffl]

@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.

[Prasanth345]

@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.

[jscheffl]

"almost" good to merge but some nit about log formatting/commented code. Then I think it is good to merge.

[jscheffl]

Thanks for the multiple iterations for the fix and staying tuned until now! Finally all is green and looking good! LGTM! Thanks!

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.