diff --git a/providers/edge/README.rst b/providers/edge/README.rst index 9bcb8112dc3a7..c10204b4749f0 100644 --- a/providers/edge/README.rst +++ b/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 `_. +in the `documentation `_. Installation ------------ @@ -60,4 +60,4 @@ PIP package Version required ================== ================== The changelog for the provider package can be found in the -`changelog `_. +`changelog `_. diff --git a/providers/edge/docs/changelog.rst b/providers/edge/docs/changelog.rst index 87759f00b20b6..393239fb395f8 100644 --- a/providers/edge/docs/changelog.rst +++ b/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 .......... diff --git a/providers/edge/provider.yaml b/providers/edge/provider.yaml index 09ce196e17823..7bb5ce88ea967 100644 --- a/providers/edge/provider.yaml +++ b/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 diff --git a/providers/edge/pyproject.toml b/providers/edge/pyproject.toml index 0e28b816ffd37..759e8c73a9cad 100644 --- a/providers/edge/pyproject.toml +++ b/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" diff --git a/providers/edge/src/airflow/providers/edge/__init__.py b/providers/edge/src/airflow/providers/edge/__init__.py index ac0be0ebf9e85..749d124f95886 100644 --- a/providers/edge/src/airflow/providers/edge/__init__.py +++ b/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" diff --git a/providers/edge/src/airflow/providers/edge/cli/api_client.py b/providers/edge/src/airflow/providers/edge/cli/api_client.py index 60c230dada1ca..c19504787d75c 100644 --- a/providers/edge/src/airflow/providers/edge/cli/api_client.py +++ b/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,13 +93,18 @@ 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) @@ -106,13 +112,18 @@ 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) diff --git a/providers/edge/src/airflow/providers/edge/get_provider_info.py b/providers/edge/src/airflow/providers/edge/get_provider_info.py index 66fa378ef7811..9f972d4371ed0 100644 --- a/providers/edge/src/airflow/providers/edge/get_provider_info.py +++ b/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",