You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "bujjibabukatta (via GitHub)" <gi...@apache.org> on 2026/06/09 08:01:34 UTC
[PR] Add deferrable=True support to SparkSubmitOperator (issue #67168) [airflow]
bujjibabukatta opened a new pull request, #68277:
URL: https://github.com/apache/airflow/pull/68277
Closes #67168
## Summary
Adds `deferrable: bool = False` parameter to `SparkSubmitOperator` as
requested in #67168 (follow-up to #67118).
## Two execution modes
| Mode | Behaviour |
|---|---|
| `deferrable=False` (default) | Existing sync path via `ResumableJobMixin` — worker slot held during polling, reconnects to existing driver on infrastructure failure |
| `deferrable=True` | Submits job then calls `self.defer()` → `SparkDriverTrigger` — worker slot freed during polling, crash recovery handled by trigger-row persistence |
## Changes
### `operators/spark_submit.py`
- Add `deferrable: bool` param (respects `operators.default_deferrable` config)
- Add `if self.deferrable:` branch in `execute()` — submits job then defers to `SparkDriverTrigger`
- Add `execute_complete()` — callback fired when trigger completes, raises `AirflowException` on failure
- Add `_build_master_rest_urls()` — builds REST API URLs from connection extras, supports HA (comma-separated masters)
### `triggers/spark_submit.py` (new)
- `SparkDriverTrigger` — async trigger that polls Spark standalone REST API via `aiohttp`
- Supports HA: tries each master URL in order on every poll
- Treats `RELAUNCHING`/`UNKNOWN` as still-alive (master in recovery)
- `serialize()` implemented so Airflow can re-create the trigger after a crash
### `provider.yaml`
- Register `triggers/spark_submit` module so Airflow triggerer process discovers `SparkDriverTrigger`
### `pyproject.toml`
- Add `aiohttp>=3.9.0` dependency (required by `SparkDriverTrigger`)
### Tests
- `tests/triggers/test_spark_submit.py` (new) — 11 tests covering `SparkDriverTrigger` serialize, run loop, HA failover, REST polling
- `tests/operators/test_spark_submit.py` — new `TestSparkSubmitOperatorDeferrable` class with 9 tests
## Relationship to #67118
Both modes share `spark_job_id` in `task_state`. A user can switch from `deferrable=False` to `deferrable=True` without any state migration.
--
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] Add deferrable=True support to SparkSubmitOperator (issue #67168) [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #68277:
URL: https://github.com/apache/airflow/pull/68277#discussion_r3384840071
##########
providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py:
##########
@@ -134,6 +134,12 @@ class SparkSubmitOperator(ResumableJobMixin, BaseOperator):
omitted, Kerberos-enabled Spark connections with both ``keytab`` and
``principal`` configured use ``requests-kerberos`` automatically.
Defaults to ``None`` (no auth for non-Kerberos connections).
+ :param deferrable: If ``True``, submits the job then defers to
+ ``SparkDriverTrigger``; the worker slot is freed while the trigger
+ polls the Spark REST API. On crash the trigger is re-created from
+ its serialised state (no reconnect needed). On user-clear, execute()
+ runs again and a fresh job is submitted.
+ If ``False`` (default), the sync ``ResumableJobMixin`` path is used.
Review Comment:
This docstring entry is a bit too verbose. I don't think there is a need to go deep into the mechanies of deferrable mode here. I understand you want to express an adjacent development i.e. `ResumableJobMixin` but I believe that might be better suited for a comment. I think you should use this instead:
`:param deferrable: Run operator in deferrable mode.`
##########
providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py:
##########
@@ -134,6 +134,12 @@ class SparkSubmitOperator(ResumableJobMixin, BaseOperator):
omitted, Kerberos-enabled Spark connections with both ``keytab`` and
``principal`` configured use ``requests-kerberos`` automatically.
Defaults to ``None`` (no auth for non-Kerberos connections).
+ :param deferrable: If ``True``, submits the job then defers to
+ ``SparkDriverTrigger``; the worker slot is freed while the trigger
+ polls the Spark REST API. On crash the trigger is re-created from
+ its serialised state (no reconnect needed). On user-clear, execute()
+ runs again and a fresh job is submitted.
+ If ``False`` (default), the sync ``ResumableJobMixin`` path is used.
Review Comment:
This docstring entry is a bit too verbose. I don't think there is a need to go deep into the mechanies of deferrable mode here. I understand you want to express an adjacent development i.e. `ResumableJobMixin` but I believe that might be better suited for a comment. I think you should use this instead:
`:param deferrable: Run operator in deferrable mode.`
--
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] Add deferrable=True support to SparkSubmitOperator (issue #67168) [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #68277:
URL: https://github.com/apache/airflow/pull/68277#discussion_r3384841133
##########
providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py:
##########
@@ -134,6 +134,12 @@ class SparkSubmitOperator(ResumableJobMixin, BaseOperator):
omitted, Kerberos-enabled Spark connections with both ``keytab`` and
``principal`` configured use ``requests-kerberos`` automatically.
Defaults to ``None`` (no auth for non-Kerberos connections).
+ :param deferrable: If ``True``, submits the job then defers to
+ ``SparkDriverTrigger``; the worker slot is freed while the trigger
+ polls the Spark REST API. On crash the trigger is re-created from
+ its serialised state (no reconnect needed). On user-clear, execute()
+ runs again and a fresh job is submitted.
+ If ``False`` (default), the sync ``ResumableJobMixin`` path is used.
Review Comment:
This docstring entry is a bit too verbose. I don't think there is a need to go deep into the mechanics of deferrable mode here. I understand you want to explain the implications of an adjacent development i.e. `ResumableJobMixin` but I believe that might be better suited for a comment. I think you should use this instead:
`:param deferrable: Run operator in deferrable mode.`
--
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] Add deferrable=True support to SparkSubmitOperator (issue #67168) [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on PR #68277:
URL: https://github.com/apache/airflow/pull/68277#issuecomment-4696793509
@bujjibabukatta
I would suggest opening a new PR if you still want to work on this. But please verify and audit any AI-generated code before submitting it for review. PRs that have too many isssues will just result in endless back and forth that becomes frustrating for everyone.
--
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] Add deferrable=True support to SparkSubmitOperator (issue #67168) [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil closed pull request #68277: Add deferrable=True support to SparkSubmitOperator (issue #67168)
URL: https://github.com/apache/airflow/pull/68277
--
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] Add deferrable=True support to SparkSubmitOperator (issue #67168) [airflow]
Posted by "bujjibabukatta (via GitHub)" <gi...@apache.org>.
bujjibabukatta commented on PR #68277:
URL: https://github.com/apache/airflow/pull/68277#issuecomment-4673437250
hi @SameerMesiah97
Thank you for the feedback! I've addressed all the comments:
1. Simplified the :param deferrable: docstring to a single line
2. Moved the SparkDriverTrigger import to the top of the file
3. Replaced AirflowException with native RuntimeError
4. Validated status and driver_id explicitly instead of using fallback defaults
5. Replaced the list comprehension in _build_master_rest_urls() with an explicit loop
--
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] Add deferrable=True support to SparkSubmitOperator (issue #67168) [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on PR #68277:
URL: https://github.com/apache/airflow/pull/68277#issuecomment-4674634713
> hi @SameerMesiah97
>
> Thank you for the feedback! I've addressed all the comments:
>
> 1. Simplified the :param deferrable: docstring to a single line
> 2. Moved the SparkDriverTrigger import to the top of the file
> 3. Replaced AirflowException with native RuntimeError
> 4. Validated status and driver_id explicitly instead of using fallback defaults
> 5. Replaced the list comprehension in _build_master_rest_urls() with an explicit loop
@bujjibabukatta
CI is still failing. And the rest of the PR has issues. I would strongly advise you to convert this to draft and review the diff again. Then once it is ready, open it for review.
--
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] Add deferrable=True support to SparkSubmitOperator (issue #67168) [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #68277:
URL: https://github.com/apache/airflow/pull/68277#discussion_r3391338967
##########
providers/apache/spark/src/airflow/providers/apache/spark/triggers/spark_submit.py:
##########
@@ -0,0 +1,110 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import asyncio
+
+from collections.abc import AsyncIterator
+from typing import Any
+
+import aiohttp
+
+from airflow.triggers.base import BaseTrigger, TriggerEvent
+
+_SPARK_TERMINAL_STATES = frozenset({"FINISHED", "FAILED", "KILLED", "ERROR"})
+_SPARK_ACTIVE_STATES = frozenset({"SUBMITTED", "RUNNING", "RELAUNCHING", "UNKNOWN"})
+
+
+class SparkDriverTrigger(BaseTrigger):
+ """
+ Async trigger that polls the Spark standalone REST API until the driver
+ reaches a terminal state. Used when SparkSubmitOperator runs with deferrable=True.
+ :param driver_id: Spark driver submission ID returned by spark-submit --rest.
+ :param master_urls: List of Spark master REST base URLs e.g. ["http://spark-master:6066"].
+ :param poll_interval: Seconds between REST API polls. Defaults to 10.
+ """
+
+ def __init__(
+ self,
+ driver_id: str,
+ master_urls: list[str],
+ poll_interval: int = 10,
+ ) -> None:
+ super().__init__()
+ self.driver_id = driver_id
+ self.master_urls = master_urls
+ self.poll_interval = poll_interval
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ return (
+ "airflow.providers.apache.spark.triggers.spark_submit.SparkDriverTrigger",
+ {
+ "driver_id": self.driver_id,
+ "master_urls": self.master_urls,
+ "poll_interval": self.poll_interval,
+ },
+ )
+
+ async def run(self) -> AsyncIterator[TriggerEvent]:
+ """Poll Spark REST API until driver reaches a terminal state."""
+ while True:
+ status = await self._poll_driver_status()
+ if status is None:
+ yield TriggerEvent(
+ {
+ "status": "error",
+ "driver_id": self.driver_id,
+ "message": "All Spark masters unreachable",
+ }
+ )
+ return
+ self.log.info("SparkDriverTrigger: driver=%s status=%s", self.driver_id, status)
+ upper = status.upper()
Review Comment:
I would change `upper` to `normalized_status` to better communicate the intent of using `upper()`
##########
providers/apache/spark/src/airflow/providers/apache/spark/triggers/spark_submit.py:
##########
@@ -0,0 +1,110 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import asyncio
+
+from collections.abc import AsyncIterator
+from typing import Any
+
+import aiohttp
+
+from airflow.triggers.base import BaseTrigger, TriggerEvent
+
+_SPARK_TERMINAL_STATES = frozenset({"FINISHED", "FAILED", "KILLED", "ERROR"})
+_SPARK_ACTIVE_STATES = frozenset({"SUBMITTED", "RUNNING", "RELAUNCHING", "UNKNOWN"})
+
+
+class SparkDriverTrigger(BaseTrigger):
+ """
+ Async trigger that polls the Spark standalone REST API until the driver
+ reaches a terminal state. Used when SparkSubmitOperator runs with deferrable=True.
+ :param driver_id: Spark driver submission ID returned by spark-submit --rest.
+ :param master_urls: List of Spark master REST base URLs e.g. ["http://spark-master:6066"].
+ :param poll_interval: Seconds between REST API polls. Defaults to 10.
+ """
+
+ def __init__(
+ self,
+ driver_id: str,
+ master_urls: list[str],
+ poll_interval: int = 10,
+ ) -> None:
+ super().__init__()
+ self.driver_id = driver_id
+ self.master_urls = master_urls
+ self.poll_interval = poll_interval
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ return (
+ "airflow.providers.apache.spark.triggers.spark_submit.SparkDriverTrigger",
+ {
+ "driver_id": self.driver_id,
+ "master_urls": self.master_urls,
+ "poll_interval": self.poll_interval,
+ },
+ )
+
+ async def run(self) -> AsyncIterator[TriggerEvent]:
+ """Poll Spark REST API until driver reaches a terminal state."""
+ while True:
+ status = await self._poll_driver_status()
+ if status is None:
+ yield TriggerEvent(
+ {
+ "status": "error",
+ "driver_id": self.driver_id,
+ "message": "All Spark masters unreachable",
+ }
+ )
+ return
+ self.log.info("SparkDriverTrigger: driver=%s status=%s", self.driver_id, status)
+ upper = status.upper()
+ if upper in _SPARK_TERMINAL_STATES:
+ success = upper == "FINISHED"
+ yield TriggerEvent(
+ {
+ "status": "success" if success else "error",
+ "driver_id": self.driver_id,
+ "driver_state": upper,
+ "message": f"Driver {self.driver_id} reached state {upper}",
+ }
+ )
+ return
+ await asyncio.sleep(self.poll_interval)
+
+ async def _poll_driver_status(self) -> str | None:
+ """Try each master URL; return driverState str or None if all fail."""
+ for url in self.master_urls:
+ status_url = f"{url.rstrip('/')}/v1/submissions/status/{self.driver_id}"
+ try:
+ async with aiohttp.ClientSession() as session:
+ async with session.get(status_url, timeout=aiohttp.ClientTimeout(total=30)) as resp:
+ resp.raise_for_status()
+ data = await resp.json()
+ if not data.get("success"):
+ self.log.warning(
+ "Spark REST API returned success=false for %s: %s",
+ self.driver_id,
+ data.get("message", "unknown"),
+ )
+ return "UNKNOWN"
Review Comment:
I am not sure why you are defaulting to `UNKNOWN` when the Spark driver status is not success? Why should the driver status always be `UNKNOWN` if not success? Also, with your current logic, the spark driver state not being retrievable will just result in the trigger polling continuously until a success state or error is encountered.
##########
providers/apache/spark/tests/unit/apache/spark/operators/test_spark_submit.py:
##########
@@ -920,3 +920,156 @@ def test_execute_falls_through_to_plain_submit_when_flag_off(self):
hook.submit.assert_called_once_with("test.jar")
hook._poll_k8s_driver_via_api.assert_not_called()
+
+
+class TestSparkSubmitOperatorDeferrable:
+ """Tests for SparkSubmitOperator deferrable=True mode."""
+
+ def setup_method(self):
+ args = {"owner": "airflow", "start_date": DEFAULT_DATE}
+ self.dag = DAG("test_deferrable_dag", schedule=None, default_args=args)
+
+ def _make_operator(self, **kwargs):
+ return SparkSubmitOperator(
+ task_id="test_deferrable",
+ dag=self.dag,
+ application="test.jar",
+ deferrable=True,
+ **kwargs,
+ )
+
+ def _make_hook(self):
+ hook = MagicMock()
+ hook._should_track_driver_status = False
+ hook._should_track_driver_via_k8s_api.return_value = False
+ hook._connection = {"master": "spark://myhost:7077"}
+ hook.submit.return_value = "driver-001"
+ return hook
+
+ def test_deferrable_defaults_to_false(self):
Review Comment:
This test is not needed as you are testing native python behaviour.
##########
providers/apache/spark/tests/unit/apache/spark/operators/test_spark_submit.py:
##########
@@ -920,3 +920,156 @@ def test_execute_falls_through_to_plain_submit_when_flag_off(self):
hook.submit.assert_called_once_with("test.jar")
hook._poll_k8s_driver_via_api.assert_not_called()
+
+
+class TestSparkSubmitOperatorDeferrable:
+ """Tests for SparkSubmitOperator deferrable=True mode."""
+
+ def setup_method(self):
+ args = {"owner": "airflow", "start_date": DEFAULT_DATE}
+ self.dag = DAG("test_deferrable_dag", schedule=None, default_args=args)
+
+ def _make_operator(self, **kwargs):
+ return SparkSubmitOperator(
+ task_id="test_deferrable",
+ dag=self.dag,
+ application="test.jar",
+ deferrable=True,
+ **kwargs,
+ )
+
+ def _make_hook(self):
+ hook = MagicMock()
+ hook._should_track_driver_status = False
+ hook._should_track_driver_via_k8s_api.return_value = False
+ hook._connection = {"master": "spark://myhost:7077"}
+ hook.submit.return_value = "driver-001"
+ return hook
+
+ def test_deferrable_defaults_to_false(self):
+ """deferrable must default to False — existing behaviour unchanged."""
+ op = SparkSubmitOperator(task_id="t", dag=self.dag, application="app.jar")
+ assert op.deferrable is False
+
+ def test_deferrable_stored_on_operator(self):
Review Comment:
Same here. Not needed.
##########
providers/apache/spark/src/airflow/providers/apache/spark/triggers/spark_submit.py:
##########
@@ -0,0 +1,110 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import asyncio
+
+from collections.abc import AsyncIterator
+from typing import Any
+
+import aiohttp
+
+from airflow.triggers.base import BaseTrigger, TriggerEvent
+
+_SPARK_TERMINAL_STATES = frozenset({"FINISHED", "FAILED", "KILLED", "ERROR"})
+_SPARK_ACTIVE_STATES = frozenset({"SUBMITTED", "RUNNING", "RELAUNCHING", "UNKNOWN"})
Review Comment:
I don't see `_SPARK_ACTIVE_STATES` being used anywhere. Why is this needed?
##########
providers/apache/spark/src/airflow/providers/apache/spark/triggers/spark_submit.py:
##########
@@ -0,0 +1,110 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import asyncio
+
+from collections.abc import AsyncIterator
+from typing import Any
+
+import aiohttp
+
+from airflow.triggers.base import BaseTrigger, TriggerEvent
+
+_SPARK_TERMINAL_STATES = frozenset({"FINISHED", "FAILED", "KILLED", "ERROR"})
Review Comment:
nit: no need for frozenset as this set is not being modified anywhere.
##########
providers/apache/spark/provider.yaml:
##########
@@ -241,3 +241,8 @@ connection-types:
task-decorators:
- class-name: airflow.providers.apache.spark.decorators.pyspark.pyspark_task
name: pyspark
+
+triggers:
Review Comment:
This should come after hooks.
##########
providers/apache/spark/src/airflow/providers/apache/spark/triggers/spark_submit.py:
##########
@@ -0,0 +1,110 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import asyncio
+
+from collections.abc import AsyncIterator
+from typing import Any
+
+import aiohttp
+
+from airflow.triggers.base import BaseTrigger, TriggerEvent
+
+_SPARK_TERMINAL_STATES = frozenset({"FINISHED", "FAILED", "KILLED", "ERROR"})
+_SPARK_ACTIVE_STATES = frozenset({"SUBMITTED", "RUNNING", "RELAUNCHING", "UNKNOWN"})
+
+
+class SparkDriverTrigger(BaseTrigger):
+ """
+ Async trigger that polls the Spark standalone REST API until the driver
+ reaches a terminal state. Used when SparkSubmitOperator runs with deferrable=True.
+ :param driver_id: Spark driver submission ID returned by spark-submit --rest.
+ :param master_urls: List of Spark master REST base URLs e.g. ["http://spark-master:6066"].
+ :param poll_interval: Seconds between REST API polls. Defaults to 10.
+ """
+
+ def __init__(
+ self,
+ driver_id: str,
+ master_urls: list[str],
+ poll_interval: int = 10,
+ ) -> None:
+ super().__init__()
+ self.driver_id = driver_id
+ self.master_urls = master_urls
+ self.poll_interval = poll_interval
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ return (
+ "airflow.providers.apache.spark.triggers.spark_submit.SparkDriverTrigger",
+ {
+ "driver_id": self.driver_id,
+ "master_urls": self.master_urls,
+ "poll_interval": self.poll_interval,
+ },
+ )
+
+ async def run(self) -> AsyncIterator[TriggerEvent]:
+ """Poll Spark REST API until driver reaches a terminal state."""
+ while True:
+ status = await self._poll_driver_status()
+ if status is None:
+ yield TriggerEvent(
+ {
+ "status": "error",
+ "driver_id": self.driver_id,
+ "message": "All Spark masters unreachable",
+ }
+ )
+ return
+ self.log.info("SparkDriverTrigger: driver=%s status=%s", self.driver_id, status)
+ upper = status.upper()
+ if upper in _SPARK_TERMINAL_STATES:
+ success = upper == "FINISHED"
+ yield TriggerEvent(
+ {
+ "status": "success" if success else "error",
+ "driver_id": self.driver_id,
+ "driver_state": upper,
+ "message": f"Driver {self.driver_id} reached state {upper}",
+ }
+ )
+ return
+ await asyncio.sleep(self.poll_interval)
+
+ async def _poll_driver_status(self) -> str | None:
+ """Try each master URL; return driverState str or None if all fail."""
+ for url in self.master_urls:
+ status_url = f"{url.rstrip('/')}/v1/submissions/status/{self.driver_id}"
+ try:
+ async with aiohttp.ClientSession() as session:
+ async with session.get(status_url, timeout=aiohttp.ClientTimeout(total=30)) as resp:
+ resp.raise_for_status()
+ data = await resp.json()
+ if not data.get("success"):
+ self.log.warning(
+ "Spark REST API returned success=false for %s: %s",
+ self.driver_id,
+ data.get("message", "unknown"),
+ )
+ return "UNKNOWN"
+ return data["driverState"]
+ except Exception as exc:
Review Comment:
This Exception is too broad. What if there is a `ClientError/TimeoutError` in the loop? It seems like you are collapsing all of these different errors into a single log message which could be misleading. Also, if the error is not recoverable, why continue the loop?
##########
providers/apache/spark/tests/unit/apache/spark/operators/test_spark_submit.py:
##########
@@ -920,3 +920,156 @@ def test_execute_falls_through_to_plain_submit_when_flag_off(self):
hook.submit.assert_called_once_with("test.jar")
hook._poll_k8s_driver_via_api.assert_not_called()
+
+
+class TestSparkSubmitOperatorDeferrable:
+ """Tests for SparkSubmitOperator deferrable=True mode."""
+
+ def setup_method(self):
+ args = {"owner": "airflow", "start_date": DEFAULT_DATE}
+ self.dag = DAG("test_deferrable_dag", schedule=None, default_args=args)
+
+ def _make_operator(self, **kwargs):
+ return SparkSubmitOperator(
+ task_id="test_deferrable",
+ dag=self.dag,
+ application="test.jar",
+ deferrable=True,
+ **kwargs,
+ )
+
+ def _make_hook(self):
+ hook = MagicMock()
+ hook._should_track_driver_status = False
+ hook._should_track_driver_via_k8s_api.return_value = False
+ hook._connection = {"master": "spark://myhost:7077"}
+ hook.submit.return_value = "driver-001"
+ return hook
+
+ def test_deferrable_defaults_to_false(self):
+ """deferrable must default to False — existing behaviour unchanged."""
+ op = SparkSubmitOperator(task_id="t", dag=self.dag, application="app.jar")
+ assert op.deferrable is False
+
+ def test_deferrable_stored_on_operator(self):
+ """deferrable=True must be stored as self.deferrable."""
+ op = self._make_operator()
+ assert op.deferrable is True
+
+ def test_execute_calls_defer_when_deferrable_true(self):
+ """execute() must call self.defer() when deferrable=True."""
+ op = self._make_operator()
+ hook = self._make_hook()
+ op._hook = hook
+
+ with mock.patch.object(op, "submit_job", return_value="driver-001"), \
+ mock.patch.object(op, "_build_master_rest_urls", return_value=["http://myhost:6066"]), \
+ mock.patch.object(op, "defer") as mock_defer:
+ op.execute(context={})
+
+ mock_defer.assert_called_once()
+ call_kwargs = mock_defer.call_args.kwargs
+ assert call_kwargs["method_name"] == "execute_complete"
+
+ def test_execute_passes_correct_args_to_trigger(self):
+ """execute() must pass driver_id and master_urls to SparkDriverTrigger."""
+ from airflow.providers.apache.spark.triggers.spark_submit import SparkDriverTrigger
+
+ op = self._make_operator(status_poll_interval=15)
+ hook = self._make_hook()
+ op._hook = hook
+
+ with mock.patch.object(op, "submit_job", return_value="driver-xyz"), \
+ mock.patch.object(op, "_build_master_rest_urls", return_value=["http://m1:6066"]), \
+ mock.patch.object(op, "defer") as mock_defer:
+ op.execute(context={})
+
+ trigger = mock_defer.call_args.kwargs["trigger"]
+ assert isinstance(trigger, SparkDriverTrigger)
+ assert trigger.driver_id == "driver-xyz"
+ assert trigger.master_urls == ["http://m1:6066"]
+ assert trigger.poll_interval == 15
+
+ def test_execute_does_not_call_hook_submit_directly(self):
+ """execute() in deferrable mode must use submit_job(), not hook.submit()."""
+ op = self._make_operator()
+ hook = self._make_hook()
+ op._hook = hook
+
+ with mock.patch.object(op, "submit_job", return_value="driver-001"), \
+ mock.patch.object(op, "_build_master_rest_urls", return_value=["http://myhost:6066"]), \
+ mock.patch.object(op, "defer"):
+ op.execute(context={})
+
+ hook.submit.assert_not_called()
+
+ def test_execute_complete_succeeds_on_success_event(self):
+ """execute_complete() must not raise when status=success."""
+ op = self._make_operator()
+ event = {
+ "status": "success",
+ "driver_id": "driver-001",
+ "driver_state": "FINISHED",
+ "message": "Driver reached FINISHED",
+ }
+ op.execute_complete(context={}, event=event) # must not raise
+
+ def test_execute_complete_raises_on_error_event(self):
+ """execute_complete() must raise AirflowException when status=error."""
+ from airflow.providers.common.compat.sdk import AirflowException
+
+ op = self._make_operator()
+ event = {
+ "status": "error",
+ "driver_id": "driver-001",
+ "driver_state": "FAILED",
+ "message": "Driver reached FAILED",
+ }
+ with pytest.raises(AirflowException, match="driver-001"):
+ op.execute_complete(context={}, event=event)
+
Review Comment:
Can you add another test for malformed event payloads i.e. missing `driver_id` and `status`?
##########
providers/apache/spark/tests/unit/apache/spark/operators/test_spark_submit.py:
##########
@@ -920,3 +920,156 @@ def test_execute_falls_through_to_plain_submit_when_flag_off(self):
hook.submit.assert_called_once_with("test.jar")
hook._poll_k8s_driver_via_api.assert_not_called()
+
+
+class TestSparkSubmitOperatorDeferrable:
+ """Tests for SparkSubmitOperator deferrable=True mode."""
+
+ def setup_method(self):
+ args = {"owner": "airflow", "start_date": DEFAULT_DATE}
+ self.dag = DAG("test_deferrable_dag", schedule=None, default_args=args)
+
+ def _make_operator(self, **kwargs):
+ return SparkSubmitOperator(
+ task_id="test_deferrable",
+ dag=self.dag,
+ application="test.jar",
+ deferrable=True,
+ **kwargs,
+ )
+
+ def _make_hook(self):
+ hook = MagicMock()
+ hook._should_track_driver_status = False
+ hook._should_track_driver_via_k8s_api.return_value = False
+ hook._connection = {"master": "spark://myhost:7077"}
+ hook.submit.return_value = "driver-001"
+ return hook
+
+ def test_deferrable_defaults_to_false(self):
+ """deferrable must default to False — existing behaviour unchanged."""
+ op = SparkSubmitOperator(task_id="t", dag=self.dag, application="app.jar")
+ assert op.deferrable is False
+
+ def test_deferrable_stored_on_operator(self):
+ """deferrable=True must be stored as self.deferrable."""
+ op = self._make_operator()
+ assert op.deferrable is True
+
+ def test_execute_calls_defer_when_deferrable_true(self):
+ """execute() must call self.defer() when deferrable=True."""
+ op = self._make_operator()
+ hook = self._make_hook()
+ op._hook = hook
+
+ with mock.patch.object(op, "submit_job", return_value="driver-001"), \
+ mock.patch.object(op, "_build_master_rest_urls", return_value=["http://myhost:6066"]), \
+ mock.patch.object(op, "defer") as mock_defer:
+ op.execute(context={})
+
+ mock_defer.assert_called_once()
+ call_kwargs = mock_defer.call_args.kwargs
+ assert call_kwargs["method_name"] == "execute_complete"
+
+ def test_execute_passes_correct_args_to_trigger(self):
+ """execute() must pass driver_id and master_urls to SparkDriverTrigger."""
+ from airflow.providers.apache.spark.triggers.spark_submit import SparkDriverTrigger
+
+ op = self._make_operator(status_poll_interval=15)
+ hook = self._make_hook()
+ op._hook = hook
+
+ with mock.patch.object(op, "submit_job", return_value="driver-xyz"), \
+ mock.patch.object(op, "_build_master_rest_urls", return_value=["http://m1:6066"]), \
+ mock.patch.object(op, "defer") as mock_defer:
+ op.execute(context={})
+
+ trigger = mock_defer.call_args.kwargs["trigger"]
+ assert isinstance(trigger, SparkDriverTrigger)
+ assert trigger.driver_id == "driver-xyz"
+ assert trigger.master_urls == ["http://m1:6066"]
+ assert trigger.poll_interval == 15
+
+ def test_execute_does_not_call_hook_submit_directly(self):
+ """execute() in deferrable mode must use submit_job(), not hook.submit()."""
+ op = self._make_operator()
+ hook = self._make_hook()
+ op._hook = hook
+
+ with mock.patch.object(op, "submit_job", return_value="driver-001"), \
+ mock.patch.object(op, "_build_master_rest_urls", return_value=["http://myhost:6066"]), \
+ mock.patch.object(op, "defer"):
+ op.execute(context={})
+
+ hook.submit.assert_not_called()
+
+ def test_execute_complete_succeeds_on_success_event(self):
+ """execute_complete() must not raise when status=success."""
+ op = self._make_operator()
+ event = {
+ "status": "success",
+ "driver_id": "driver-001",
+ "driver_state": "FINISHED",
+ "message": "Driver reached FINISHED",
+ }
+ op.execute_complete(context={}, event=event) # must not raise
+
+ def test_execute_complete_raises_on_error_event(self):
+ """execute_complete() must raise AirflowException when status=error."""
+ from airflow.providers.common.compat.sdk import AirflowException
Review Comment:
Why is `AirflowException` still here? You switched to `RuntimeError` in your implementation?
##########
providers/apache/spark/tests/unit/apache/spark/operators/test_spark_submit.py:
##########
@@ -920,3 +920,156 @@ def test_execute_falls_through_to_plain_submit_when_flag_off(self):
hook.submit.assert_called_once_with("test.jar")
hook._poll_k8s_driver_via_api.assert_not_called()
+
+
+class TestSparkSubmitOperatorDeferrable:
+ """Tests for SparkSubmitOperator deferrable=True mode."""
+
+ def setup_method(self):
+ args = {"owner": "airflow", "start_date": DEFAULT_DATE}
+ self.dag = DAG("test_deferrable_dag", schedule=None, default_args=args)
+
+ def _make_operator(self, **kwargs):
+ return SparkSubmitOperator(
+ task_id="test_deferrable",
+ dag=self.dag,
+ application="test.jar",
+ deferrable=True,
+ **kwargs,
+ )
+
+ def _make_hook(self):
+ hook = MagicMock()
+ hook._should_track_driver_status = False
+ hook._should_track_driver_via_k8s_api.return_value = False
+ hook._connection = {"master": "spark://myhost:7077"}
+ hook.submit.return_value = "driver-001"
+ return hook
+
+ def test_deferrable_defaults_to_false(self):
+ """deferrable must default to False — existing behaviour unchanged."""
+ op = SparkSubmitOperator(task_id="t", dag=self.dag, application="app.jar")
+ assert op.deferrable is False
+
+ def test_deferrable_stored_on_operator(self):
+ """deferrable=True must be stored as self.deferrable."""
+ op = self._make_operator()
+ assert op.deferrable is True
+
+ def test_execute_calls_defer_when_deferrable_true(self):
+ """execute() must call self.defer() when deferrable=True."""
+ op = self._make_operator()
+ hook = self._make_hook()
+ op._hook = hook
+
+ with mock.patch.object(op, "submit_job", return_value="driver-001"), \
+ mock.patch.object(op, "_build_master_rest_urls", return_value=["http://myhost:6066"]), \
+ mock.patch.object(op, "defer") as mock_defer:
+ op.execute(context={})
+
+ mock_defer.assert_called_once()
+ call_kwargs = mock_defer.call_args.kwargs
+ assert call_kwargs["method_name"] == "execute_complete"
+
+ def test_execute_passes_correct_args_to_trigger(self):
+ """execute() must pass driver_id and master_urls to SparkDriverTrigger."""
+ from airflow.providers.apache.spark.triggers.spark_submit import SparkDriverTrigger
+
+ op = self._make_operator(status_poll_interval=15)
+ hook = self._make_hook()
+ op._hook = hook
+
+ with mock.patch.object(op, "submit_job", return_value="driver-xyz"), \
+ mock.patch.object(op, "_build_master_rest_urls", return_value=["http://m1:6066"]), \
+ mock.patch.object(op, "defer") as mock_defer:
+ op.execute(context={})
+
+ trigger = mock_defer.call_args.kwargs["trigger"]
+ assert isinstance(trigger, SparkDriverTrigger)
+ assert trigger.driver_id == "driver-xyz"
+ assert trigger.master_urls == ["http://m1:6066"]
+ assert trigger.poll_interval == 15
+
+ def test_execute_does_not_call_hook_submit_directly(self):
+ """execute() in deferrable mode must use submit_job(), not hook.submit()."""
+ op = self._make_operator()
+ hook = self._make_hook()
+ op._hook = hook
+
+ with mock.patch.object(op, "submit_job", return_value="driver-001"), \
+ mock.patch.object(op, "_build_master_rest_urls", return_value=["http://myhost:6066"]), \
+ mock.patch.object(op, "defer"):
+ op.execute(context={})
+
+ hook.submit.assert_not_called()
+
+ def test_execute_complete_succeeds_on_success_event(self):
+ """execute_complete() must not raise when status=success."""
+ op = self._make_operator()
+ event = {
+ "status": "success",
+ "driver_id": "driver-001",
+ "driver_state": "FINISHED",
+ "message": "Driver reached FINISHED",
+ }
+ op.execute_complete(context={}, event=event) # must not raise
+
+ def test_execute_complete_raises_on_error_event(self):
+ """execute_complete() must raise AirflowException when status=error."""
+ from airflow.providers.common.compat.sdk import AirflowException
+
+ op = self._make_operator()
+ event = {
+ "status": "error",
+ "driver_id": "driver-001",
+ "driver_state": "FAILED",
+ "message": "Driver reached FAILED",
+ }
+ with pytest.raises(AirflowException, match="driver-001"):
+ op.execute_complete(context={}, event=event)
+
+ def test_build_master_rest_urls_single_master(self):
+ """_build_master_rest_urls must return correct URL for a single master."""
+ op = self._make_operator()
+ hook = self._make_hook()
+ hook._connection = {
+ "master": "spark://myhost:7077",
+ "rest_scheme": "http",
+ "rest_port": 6066,
+ }
+ op._hook = hook
+
+ urls = op._build_master_rest_urls()
+
+ assert urls == ["http://myhost:6066"]
+
+ def test_build_master_rest_urls_ha_multiple_masters(self):
+ """_build_master_rest_urls must return a URL per master in HA mode."""
+ op = self._make_operator()
+ hook = self._make_hook()
+ hook._connection = {
+ "master": "spark://m1:7077,m2:7077",
+ "rest_scheme": "https",
+ "rest_port": 6066,
+ }
+ op._hook = hook
+
+ urls = op._build_master_rest_urls()
+
+ assert urls == ["https://m1:6066", "https://m2:6066"]
+
+ def test_deferrable_false_uses_sync_path(self):
Review Comment:
Is this not covered by the existing non-deferrable tests?
--
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] Add deferrable=True support to SparkSubmitOperator (issue #67168) [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #68277:
URL: https://github.com/apache/airflow/pull/68277#discussion_r3384904879
##########
providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py:
##########
@@ -275,6 +296,36 @@ def execute(self, context: Context) -> None:
hook._poll_k8s_driver_via_api()
return
hook.submit(self.application)
+
+ def execute_complete(self, context: Context, event: dict) -> None:
+ """
+ Handle the result emitted by SparkDriverTrigger.
+ Called by Airflow when the trigger fires after deferrable=True execution.
+ Raises AirflowException if the driver did not finish successfully.
+ """
+ from airflow.providers.common.compat.sdk import AirflowException
+ driver_state = event.get("driver_state", "UNKNOWN")
+ driver_id = event.get("driver_id", "unknown")
+ message = event.get("message", "")
+ if event.get("status") != "success":
+ raise AirflowException(
+ f"Spark driver {driver_id} did not finish successfully "
+ f"(state={driver_state}): {message}"
+ )
+ self.log.info("Spark driver %s finished successfully (state=%s)", driver_id, driver_state)
+ def _build_master_rest_urls(self) -> list[str]:
+ """
+ Build Spark master REST API base URLs for SparkDriverTrigger.
+ Supports HA (comma-separated master URL) and respects rest_scheme /
+ rest_port connection extras (same logic as get_job_status).
+ """
+ if self._hook is None:
+ self._hook = self._get_hook()
+ scheme = self._hook._connection.get("rest_scheme", "http")
+ rest_port = self._hook._connection.get("rest_port", 6066)
+ master_hosts = self._hook._connection["master"].replace("spark://", "").split(",")
+ return [f"{scheme}://{m.strip().split(':')[0]}:{rest_port}" for m in master_hosts]
Review Comment:
I think the list comprehension is doing way too much heavy lifting. An explicit loop like this would be better:
```
urls = []
for host in master_hosts:
hostname = host.strip().split(":")[0]
urls.append(f"{scheme}://{hostname}:{rest_port}")
return urls
```
You can see that it is much easier to understand the string parsing logic.
##########
providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py:
##########
@@ -275,6 +296,36 @@ def execute(self, context: Context) -> None:
hook._poll_k8s_driver_via_api()
return
hook.submit(self.application)
+
+ def execute_complete(self, context: Context, event: dict) -> None:
+ """
+ Handle the result emitted by SparkDriverTrigger.
+ Called by Airflow when the trigger fires after deferrable=True execution.
+ Raises AirflowException if the driver did not finish successfully.
+ """
+ from airflow.providers.common.compat.sdk import AirflowException
+ driver_state = event.get("driver_state", "UNKNOWN")
+ driver_id = event.get("driver_id", "unknown")
Review Comment:
So I can see 2 issues here:
1) The fallback for `driver_state` when the event payload containts no "driver_state" is defaulting to `UNKNOWN` which is a valid spark driver state. Is it really the same as the state not being retrievable?
2) There is an overlap between the fallback value for `driver_id` and `UNKNOWN`, which may confuse users reading the logs.
I think it would be better to validate both like this:
```
status = event.get("status")
driver_id = event.get("driver_id")
if status is None:
raise RuntimeError(f"Malformed trigger event: {event}")
if driver_id is None:
raise RuntimeError(f"Malformed trigger event: {event}")
```
##########
providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py:
##########
@@ -261,6 +269,19 @@ def execute(self, context: Context) -> None:
hook = self._hook
if self._track_driver_via_k8s_api:
hook._validate_track_driver_via_k8s_api_config()
+ if self.deferrable:
+ driver_id = self.submit_job(context)
+ master_urls = self._build_master_rest_urls()
+ from airflow.providers.apache.spark.triggers.spark_submit import SparkDriverTrigger
Review Comment:
This import belongs at the top of the file.
##########
providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py:
##########
@@ -275,6 +296,36 @@ def execute(self, context: Context) -> None:
hook._poll_k8s_driver_via_api()
return
hook.submit(self.application)
+
+ def execute_complete(self, context: Context, event: dict) -> None:
+ """
+ Handle the result emitted by SparkDriverTrigger.
+ Called by Airflow when the trigger fires after deferrable=True execution.
+ Raises AirflowException if the driver did not finish successfully.
+ """
+ from airflow.providers.common.compat.sdk import AirflowException
+ driver_state = event.get("driver_state", "UNKNOWN")
+ driver_id = event.get("driver_id", "unknown")
+ message = event.get("message", "")
+ if event.get("status") != "success":
+ raise AirflowException(
+ f"Spark driver {driver_id} did not finish successfully "
+ f"(state={driver_state}): {message}"
+ )
+ self.log.info("Spark driver %s finished successfully (state=%s)", driver_id, driver_state)
+ def _build_master_rest_urls(self) -> list[str]:
+ """
+ Build Spark master REST API base URLs for SparkDriverTrigger.
+ Supports HA (comma-separated master URL) and respects rest_scheme /
+ rest_port connection extras (same logic as get_job_status).
+ """
+ if self._hook is None:
+ self._hook = self._get_hook()
+ scheme = self._hook._connection.get("rest_scheme", "http")
Review Comment:
I am not sure why you are accessing the private attribute `_connection` to get the scheme. Isn't there a public API you can use?
##########
providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py:
##########
@@ -134,6 +134,12 @@ class SparkSubmitOperator(ResumableJobMixin, BaseOperator):
omitted, Kerberos-enabled Spark connections with both ``keytab`` and
``principal`` configured use ``requests-kerberos`` automatically.
Defaults to ``None`` (no auth for non-Kerberos connections).
+ :param deferrable: If ``True``, submits the job then defers to
+ ``SparkDriverTrigger``; the worker slot is freed while the trigger
+ polls the Spark REST API. On crash the trigger is re-created from
+ its serialised state (no reconnect needed). On user-clear, execute()
+ runs again and a fresh job is submitted.
+ If ``False`` (default), the sync ``ResumableJobMixin`` path is used.
Review Comment:
This docstring entry is a bit too verbose. I don't think there is a need to go deep into the mechanies of deferrable mode here. I understand you want to explain the implications of an adjacent development i.e. `ResumableJobMixin` but I believe that might be better suited for a comment. I think you should use this instead:
:param deferrable: Run operator in deferrable mode.
##########
providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py:
##########
@@ -275,6 +296,36 @@ def execute(self, context: Context) -> None:
hook._poll_k8s_driver_via_api()
return
hook.submit(self.application)
+
+ def execute_complete(self, context: Context, event: dict) -> None:
+ """
+ Handle the result emitted by SparkDriverTrigger.
+ Called by Airflow when the trigger fires after deferrable=True execution.
+ Raises AirflowException if the driver did not finish successfully.
+ """
+ from airflow.providers.common.compat.sdk import AirflowException
Review Comment:
Same here. And there has been a consensus for a while to move away from `AirflowException` in favour of native python Errors for e.g. `ValueError`, `RuntineError` or SDK exceptions if available. I would remove `AirflowException`.
--
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