Fix edge in rare conditions that task state can not be reported as cl…

…eaned in parallel
apache · Nov 23, 2024 · 39f6237 · 39f6237
1 parent b1e0212
commit 39f6237
Show file tree

Hide file tree

Showing 4 changed files with 14 additions and 5 deletions.
diff --git a/providers/src/airflow/providers/edge/CHANGELOG.rst b/providers/src/airflow/providers/edge/CHANGELOG.rst
@@ -27,6 +27,14 @@
 Changelog
 ---------
 
+0.6.2pre0
+.........
+
+Misc
+~~~~
+
+* ``Fix race that reporting status fails if the task has been cleaned in parallel.``
+
 0.6.1pre0
 .........
 

diff --git a/providers/src/airflow/providers/edge/__init__.py b/providers/src/airflow/providers/edge/__init__.py
@@ -29,7 +29,7 @@
 
 __all__ = ["__version__"]
 
-__version__ = "0.6.1pre0"
+__version__ = "0.6.2pre0"
 
 if packaging.version.parse(packaging.version.parse(airflow_version).base_version) < packaging.version.parse(
     "2.10.0"

diff --git a/providers/src/airflow/providers/edge/models/edge_job.py b/providers/src/airflow/providers/edge/models/edge_job.py
@@ -172,9 +172,10 @@ def set_state(task: TaskInstanceKey | tuple, state: TaskInstanceState, session:
             EdgeJobModel.try_number == task.try_number,
         )
         job: EdgeJobModel = session.scalar(query)
-        job.state = state
-        job.last_update = timezone.utcnow()
-        session.commit()
+        if job:
+            job.state = state
+            job.last_update = timezone.utcnow()
+            session.commit()
 
     def __hash__(self):
         return f"{self.dag_id}|{self.task_id}|{self.run_id}|{self.map_index}|{self.try_number}".__hash__()

diff --git a/providers/src/airflow/providers/edge/provider.yaml b/providers/src/airflow/providers/edge/provider.yaml
@@ -27,7 +27,7 @@ source-date-epoch: 1729683247
 
 # note that those versions are maintained by release manager - do not update them manually
 versions:
-  - 0.6.1pre0
+  - 0.6.2pre0
 
 dependencies:
   - apache-airflow>=2.10.0