Bugfix/edge worker restarts on version mismatch

providers/edge/README.rst

@@ -24,7 +24,7 @@
 
 Package ``apache-airflow-providers-edge``
 
-Release: ``0.13.0pre0``
+Release: ``0.13.1pre0``
 
 
 Handle edge workers on remote sites via HTTP(s) connection and orchestrates work over distributed sites
@@ -37,7 +37,7 @@ This is a provider package for ``edge`` provider. All classes for this provider
 are in ``airflow.providers.edge`` python package.
 
 You can find package information and changelog for the provider
-in the `documentation <https://airflow.apache.org/docs/apache-airflow-providers-edge/0.13.0pre0/>`_.
+in the `documentation <https://airflow.apache.org/docs/apache-airflow-providers-edge/0.13.1pre0/>`_.
 
 Installation
 ------------
@@ -60,4 +60,4 @@ PIP package         Version required
 ==================  ==================
 
 The changelog for the provider package can be found in the
-`changelog <https://airflow.apache.org/docs/apache-airflow-providers-edge/0.13.0pre0/changelog.html>`_.
+`changelog <https://airflow.apache.org/docs/apache-airflow-providers-edge/0.13.1pre0/changelog.html>`_.

providers/edge/docs/changelog.rst

@@ -27,6 +27,14 @@
 Changelog
 ---------
 
+0.13.1pre0
+..........
+
+Fix
+~~~
+
+* ``EdgeWorkerVersionException is raised if http 400 is responded on set_state.``
+
 0.13.0pre0
 ..........
 

providers/edge/provider.yaml

@@ -25,7 +25,7 @@ source-date-epoch: 1737371680
 
 # note that those versions are maintained by release manager - do not update them manually
 versions:
-  - 0.13.0pre0
+  - 0.13.1pre0
 
 plugins:
   - name: edge_executor

providers/edge/pyproject.toml

@@ -25,7 +25,7 @@ build-backend = "flit_core.buildapi"
 
 [project]
 name = "apache-airflow-providers-edge"
-version = "0.13.0pre0"
+version = "0.13.1pre0"
 description = "Provider package apache-airflow-providers-edge for Apache Airflow"
 readme = "README.rst"
 authors = [
@@ -62,8 +62,8 @@ dependencies = [
 ]
 
 [project.urls]
-"Documentation" = "https://airflow.apache.org/docs/apache-airflow-providers-edge/0.13.0pre0"
-"Changelog" = "https://airflow.apache.org/docs/apache-airflow-providers-edge/0.13.0pre0/changelog.html"
+"Documentation" = "https://airflow.apache.org/docs/apache-airflow-providers-edge/0.13.1pre0"
+"Changelog" = "https://airflow.apache.org/docs/apache-airflow-providers-edge/0.13.1pre0/changelog.html"
 "Bug Tracker" = "https://github.com/apache/airflow/issues"
 "Source Code" = "https://github.com/apache/airflow"
 "Slack Chat" = "https://s.apache.org/airflow-slack"

providers/edge/src/airflow/providers/edge/__init__.py

@@ -29,7 +29,7 @@
 
 __all__ = ["__version__"]
 
-__version__ = "0.13.0pre0"
+__version__ = "0.13.1pre0"
 
 if packaging.version.parse(packaging.version.parse(airflow_version).base_version) < packaging.version.parse(
     "2.10.0"

providers/edge/src/airflow/providers/edge/cli/api_client.py

@@ -30,6 +30,7 @@
 from tenacity import before_log, wait_random_exponential
 
 from airflow.configuration import conf
+from airflow.providers.edge.models.edge_worker import EdgeWorkerVersionException
 from airflow.providers.edge.worker_api.auth import jwt_signer
 from airflow.providers.edge.worker_api.datamodels import (
     EdgeJobFetched,
@@ -92,27 +93,37 @@ def worker_register(
     hostname: str, state: EdgeWorkerState, queues: list[str] | None, sysinfo: dict
 ) -> datetime:
     """Register worker with the Edge API."""
-    result = _make_generic_request(
-        "POST",
-        f"worker/{quote(hostname)}",
-        WorkerStateBody(state=state, jobs_active=0, queues=queues, sysinfo=sysinfo).model_dump_json(
-            exclude_unset=True
-        ),
-    )
+    try:
+        result = _make_generic_request(
+            "POST",
+            f"worker/{quote(hostname)}",
+            WorkerStateBody(state=state, jobs_active=0, queues=queues, sysinfo=sysinfo).model_dump_json(
+                exclude_unset=True
+            ),
+        )
+    except requests.HTTPError as e:
+        if e.response.status_code == 400:
+            raise EdgeWorkerVersionException(str(e))
+        raise e
     return datetime.fromisoformat(result)
 
 
 def worker_set_state(
     hostname: str, state: EdgeWorkerState, jobs_active: int, queues: list[str] | None, sysinfo: dict
 ) -> WorkerSetStateReturn:
     """Update the state of the worker in the central site and thereby implicitly heartbeat."""
-    result = _make_generic_request(
-        "PATCH",
-        f"worker/{quote(hostname)}",
-        WorkerStateBody(state=state, jobs_active=jobs_active, queues=queues, sysinfo=sysinfo).model_dump_json(
-            exclude_unset=True
-        ),
-    )
+    try:
+        result = _make_generic_request(
+            "PATCH",
+            f"worker/{quote(hostname)}",
+            WorkerStateBody(
+                state=state, jobs_active=jobs_active, queues=queues, sysinfo=sysinfo
+            ).model_dump_json(exclude_unset=True),
+        )
+    except requests.HTTPError as e:
+        if e.response.status_code == 400:
+            raise EdgeWorkerVersionException(str(e))
+        raise e
     return WorkerSetStateReturn(**result)
 
 

providers/edge/src/airflow/providers/edge/get_provider_info.py

@@ -28,7 +28,7 @@ def get_provider_info():
         "description": "Handle edge workers on remote sites via HTTP(s) connection and orchestrates work over distributed sites\n",
         "state": "not-ready",
         "source-date-epoch": 1737371680,
-        "versions": ["0.13.0pre0"],
+        "versions": ["0.13.1pre0"],
         "plugins": [
             {
                 "name": "edge_executor",