You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "raushanprabhakar1 (via GitHub)" <gi...@apache.org> on 2026/04/23 19:15:50 UTC
[PR] Cancel Dataproc submit triggers on user kill via on_kill() [airflow]
raushanprabhakar1 opened a new pull request, #65743:
URL: https://github.com/apache/airflow/pull/65743
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
`DataprocSubmitTrigger` and `DataprocSubmitJobDirectTrigger` now cancel in-flight Dataproc jobs via `BaseTrigger.on_kill()` when the user kills the deferred task, instead of catching `asyncio.CancelledError` in `run()` and using `safe_to_cancel()` to guess whether the triggerer was restarting. Polling in `run()` is unchanged aside from no longer handling cancellation there; `cancel_on_kill` and `job_id` still gate whether `cancel_job` runs.
Unit tests assert that `CancelledError` during polling does not call `cancel_job`, and that `on_kill()` performs the cancel (or skips when `cancel_on_kill` is false or `job_id` is missing).
related: #65733
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
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] Cancel Dataproc submit triggers on user kill via on_kill() [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #65743:
URL: https://github.com/apache/airflow/pull/65743#discussion_r3136636407
##########
providers/google/src/airflow/providers/google/cloud/triggers/dataproc.py:
##########
@@ -123,111 +123,31 @@ def serialize(self):
},
)
- if not AIRFLOW_V_3_0_PLUS:
Review Comment:
I don't understand why you removed this. This will break cancellation semantics for this trigger for the Airflow 3.x versions before 3.30 (which appears to be the version where `on_kill()` is being introduced in `BaseTrigger`).
##########
providers/google/src/airflow/providers/google/cloud/triggers/dataproc.py:
##########
@@ -123,111 +123,31 @@ def serialize(self):
},
)
- if not AIRFLOW_V_3_0_PLUS:
-
- @provide_session
- def get_task_instance(self, session: Session) -> TaskInstance:
- """
- Get the task instance for the current task.
-
- :param session: Sqlalchemy session
- """
- task_instance = session.scalar(
- select(TaskInstance).where(
- TaskInstance.dag_id == self.task_instance.dag_id,
- TaskInstance.task_id == self.task_instance.task_id,
- TaskInstance.run_id == self.task_instance.run_id,
- TaskInstance.map_index == self.task_instance.map_index,
- )
- )
- if task_instance is None:
- raise AirflowException(
- "TaskInstance with dag_id: %s,task_id: %s, run_id: %s and map_index: %s is not found",
- self.task_instance.dag_id,
- self.task_instance.task_id,
- self.task_instance.run_id,
- self.task_instance.map_index,
- )
- return task_instance
-
- async def get_task_state(self):
- from airflow.sdk.execution_time.task_runner import RuntimeTaskInstance
-
- task_states_response = await sync_to_async(RuntimeTaskInstance.get_task_states)(
- dag_id=self.task_instance.dag_id,
- task_ids=[self.task_instance.task_id],
- run_ids=[self.task_instance.run_id],
- map_index=self.task_instance.map_index,
+ async def on_kill(self) -> None:
+ """Cancel the Dataproc job when the trigger is cancelled by a user action."""
+ if not (self.job_id and self.cancel_on_kill):
+ return
+ self.log.info("Cancelling the job: %s", self.job_id)
+ await sync_to_async(self.get_sync_hook().cancel_job)(
+ job_id=self.job_id, project_id=self.project_id, region=self.region
)
- try:
- task_state = task_states_response[self.task_instance.run_id][self.task_instance.task_id]
- except Exception:
- raise AirflowException(
- "TaskInstance with dag_id: %s, task_id: %s, run_id: %s and map_index: %s is not found",
- self.task_instance.dag_id,
- self.task_instance.task_id,
- self.task_instance.run_id,
- self.task_instance.map_index,
- )
- return task_state
-
- async def safe_to_cancel(self) -> bool:
- """
- Whether it is safe to cancel the external job which is being executed by this trigger.
-
- This is to avoid the case that `asyncio.CancelledError` is called because the trigger itself is stopped.
- Because in those cases, we should NOT cancel the external job.
- """
- if AIRFLOW_V_3_0_PLUS:
- task_state = await self.get_task_state()
- else:
- # Database query is needed to get the latest state of the task instance.
- task_instance = self.get_task_instance() # type: ignore[call-arg]
- task_state = task_instance.state
- return task_state != TaskInstanceState.DEFERRED
Review Comment:
You’re removing quite a bit of existing code here, which introduces a backwards compatibility issue for pre-3.30 Airflow. In those versions `on_kill()` isn’t invoked, so cancellation would no longer go through. I’d have another look at PR #65672; there `on_kill()` is added without removing the existing `CancelledError` path.
I think it would be better to just add `on_kill()` (with tests) and leave the current logic in `run() `as-is. That way older versions continue to use the existing code path, and newer versions can make use of `on_kill()` when it’s available.
--
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] Cancel Dataproc submit triggers on user kill via on_kill() [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 closed pull request #65743: Cancel Dataproc submit triggers on user kill via on_kill()
URL: https://github.com/apache/airflow/pull/65743
--
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] Cancel Dataproc submit triggers on user kill via on_kill() [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 commented on PR #65743:
URL: https://github.com/apache/airflow/pull/65743#issuecomment-4338942978
> Looks like this has already been implemented via PR #65742 which has been merged.
Closing due to the above.
Thanks for your contribution - I hope that your next PR will be merged!
--
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] Cancel Dataproc submit triggers on user kill via on_kill() [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on PR #65743:
URL: https://github.com/apache/airflow/pull/65743#issuecomment-4338663386
Looks like this has already been implemented via PR #65742 which has been merged.
--
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] Cancel Dataproc submit triggers on user kill via on_kill() [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #65743:
URL: https://github.com/apache/airflow/pull/65743#issuecomment-4331059544
@raushanprabhakar1 This PR has a few issues that need to be addressed before it can be reviewed — please see our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
**Issues found:**
- :x: **Pre-commit / static checks**: Failing: `CI image checks / Static checks`. Run `prek run --from-ref main --stage pre-commit` locally to reproduce and fix the offending hooks.
**What to do next:**
- Fix each issue listed above.
- Make sure static checks pass locally (`prek run --from-ref main --stage pre-commit`).
- Mark the PR as "Ready for review" when you're done.
There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the [Airflow Slack](https://s.apache.org/airflow-slack).
---
_Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this [two-stage triage process](https://github.com/apache/airflow/blob/main/contributing-docs/25_maintainer_pr_triage.md#why-the-first-pass-is-automated) so that our maintainers' limited time is spent where it matters most: the conversation with you._
--
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