Add `on_kill` to Databricks Workflow Operator by R7L208 · Pull Request #42115 · apache/airflow

R7L208 · 2024-09-09T16:25:10Z

The Databricks Provider did not implement on_kill to cancel tasks generated by _CreateDatabricksWorkflowOperator. This led to data quality issues, where Airflow would report a cancellation due to timeout; however, the corresponding workflow task would continue to run on Databricks.

This PR implements on_kill for _CreateDatabricksWorkflowOperator.

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

boring-cyborg · 2024-09-09T16:25:15Z

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:

Pay attention to the quality of your code (ruff, mypy and type annotations). Our pre-commits will help you with that.
In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
Be sure to read the Airflow Coding style.
Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack

R7L208 · 2024-10-01T23:57:42Z

hey @pankajkoti - Any ETA on when you'd be able to review this PR? 🙏

pankajkoti

hi @R7L208 Apologies, I missed reviewing it so far.

I am happy to review the changes for databricks_workflow.py but not so comfortable yet reviewing the changes in databricks_sql.py (operators & hooks) as I do not have enough expertise on the changes made there. Would you please to separate out the two changes(move out databricks_sql.py related changes to another PR) and then we can invite some expert to review the other PR?

R7L208 · 2024-10-02T14:43:45Z

Opened #42668

@pankajkoti @eladkal - can you please assign someone to review?

pankajkoti · 2024-10-02T15:09:30Z

@R7L208 Can you please update the PR description based on the altered scope of this PR?

R7L208 · 2024-10-02T17:36:03Z

@pankajkoti - it's been updated

…rkflows.py

pankajkoti · 2024-10-02T17:45:26Z

@pankajkoti - it's been updated

@R7L208 it's not. It mentions about DatabricksSqlHook queries ("SQL queries submitted by DatabricksSqlHook") & using threading ("uses threading to cancel SQL queries submitted by DatabricksSqlHook.run()") which we no longer have in this PR. Could you please re-read the description and update to what's limited to the scope of this PR.

Also, _CreateDatabricksWorkflowOperator does not rely on DatabricksSqlHook, but leverages DatabricksHook just in case you missed checking that :)

R7L208 · 2024-10-02T17:50:58Z

@pankajkoti - Apologies! I read PR title instead of PR description 🤦

PR description is now updated

boring-cyborg · 2024-10-03T12:36:32Z

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.

* add on_kill override to databricks workflow operator * on_kill equivalent for DatabricksSqlOperator * add tests for create_timeout_thread * add note for on_kill in DatabricksCopyIntoOperator * chore: static checks * remove changes for databricks_sql.py for PR isolated to databricks_workflows.py --------- Co-authored-by: Lorin <lorin@databricks.com>

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed. DatabricksTaskBaseOperator — used by DatabricksTaskOperator and DatabricksNotebookOperator — is missing the same implementation, so Databricks jobs continue running after the Airflow task is killed, orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. Uses self.databricks_run_id which is initialised to None in __init__ and set by _launch_job() the moment the run is submitted.

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, and workflow-member regression (parent run_id=1, child run_id=999, asserts cancel_run(999)).

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

…ll (#69442) DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in #42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

…ll (apache#69442) DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

R7L208 · 2024-09-09T16:25:10Z

The Databricks Provider did not implement on_kill to cancel tasks generated by _CreateDatabricksWorkflowOperator. This led to data quality issues, where Airflow would report a cancellation due to timeout; however, the corresponding workflow task would continue to run on Databricks.

This PR implements on_kill for _CreateDatabricksWorkflowOperator.

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

boring-cyborg · 2024-09-09T16:25:15Z

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:

Pay attention to the quality of your code (ruff, mypy and type annotations). Our pre-commits will help you with that.
In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
Be sure to read the Airflow Coding style.
Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack

R7L208 · 2024-10-01T23:57:42Z

hey @pankajkoti - Any ETA on when you'd be able to review this PR? 🙏

pankajkoti

hi @R7L208 Apologies, I missed reviewing it so far.

I am happy to review the changes for databricks_workflow.py but not so comfortable yet reviewing the changes in databricks_sql.py (operators & hooks) as I do not have enough expertise on the changes made there. Would you please to separate out the two changes(move out databricks_sql.py related changes to another PR) and then we can invite some expert to review the other PR?

R7L208 · 2024-10-02T14:43:45Z

Opened #42668

@pankajkoti @eladkal - can you please assign someone to review?

pankajkoti · 2024-10-02T15:09:30Z

@R7L208 Can you please update the PR description based on the altered scope of this PR?

R7L208 · 2024-10-02T17:36:03Z

@pankajkoti - it's been updated

…rkflows.py

pankajkoti · 2024-10-02T17:45:26Z

@pankajkoti - it's been updated

@R7L208 it's not. It mentions about DatabricksSqlHook queries ("SQL queries submitted by DatabricksSqlHook") & using threading ("uses threading to cancel SQL queries submitted by DatabricksSqlHook.run()") which we no longer have in this PR. Could you please re-read the description and update to what's limited to the scope of this PR.

Also, _CreateDatabricksWorkflowOperator does not rely on DatabricksSqlHook, but leverages DatabricksHook just in case you missed checking that :)

R7L208 · 2024-10-02T17:50:58Z

@pankajkoti - Apologies! I read PR title instead of PR description 🤦

PR description is now updated

boring-cyborg · 2024-10-03T12:36:32Z

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.

* add on_kill override to databricks workflow operator * on_kill equivalent for DatabricksSqlOperator * add tests for create_timeout_thread * add note for on_kill in DatabricksCopyIntoOperator * chore: static checks * remove changes for databricks_sql.py for PR isolated to databricks_workflows.py --------- Co-authored-by: Lorin <lorin@databricks.com>

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed. DatabricksTaskBaseOperator — used by DatabricksTaskOperator and DatabricksNotebookOperator — is missing the same implementation, so Databricks jobs continue running after the Airflow task is killed, orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. Uses self.databricks_run_id which is initialised to None in __init__ and set by _launch_job() the moment the run is submitted.

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, and workflow-member regression (parent run_id=1, child run_id=999, asserts cancel_run(999)).

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

…ll (#69442) DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in #42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

…ll (apache#69442) DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

R7L208 · 2024-09-09T16:25:10Z

The Databricks Provider did not implement on_kill to cancel tasks generated by _CreateDatabricksWorkflowOperator. This led to data quality issues, where Airflow would report a cancellation due to timeout; however, the corresponding workflow task would continue to run on Databricks.

This PR implements on_kill for _CreateDatabricksWorkflowOperator.

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

boring-cyborg · 2024-09-09T16:25:15Z

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:

Pay attention to the quality of your code (ruff, mypy and type annotations). Our pre-commits will help you with that.
In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
Be sure to read the Airflow Coding style.
Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack

R7L208 · 2024-10-01T23:57:42Z

hey @pankajkoti - Any ETA on when you'd be able to review this PR? 🙏

pankajkoti

hi @R7L208 Apologies, I missed reviewing it so far.

I am happy to review the changes for databricks_workflow.py but not so comfortable yet reviewing the changes in databricks_sql.py (operators & hooks) as I do not have enough expertise on the changes made there. Would you please to separate out the two changes(move out databricks_sql.py related changes to another PR) and then we can invite some expert to review the other PR?

R7L208 · 2024-10-02T14:43:45Z

Opened #42668

@pankajkoti @eladkal - can you please assign someone to review?

pankajkoti · 2024-10-02T15:09:30Z

@R7L208 Can you please update the PR description based on the altered scope of this PR?

R7L208 · 2024-10-02T17:36:03Z

@pankajkoti - it's been updated

…rkflows.py

pankajkoti · 2024-10-02T17:45:26Z

@pankajkoti - it's been updated

@R7L208 it's not. It mentions about DatabricksSqlHook queries ("SQL queries submitted by DatabricksSqlHook") & using threading ("uses threading to cancel SQL queries submitted by DatabricksSqlHook.run()") which we no longer have in this PR. Could you please re-read the description and update to what's limited to the scope of this PR.

Also, _CreateDatabricksWorkflowOperator does not rely on DatabricksSqlHook, but leverages DatabricksHook just in case you missed checking that :)

R7L208 · 2024-10-02T17:50:58Z

@pankajkoti - Apologies! I read PR title instead of PR description 🤦

PR description is now updated

boring-cyborg · 2024-10-03T12:36:32Z

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.

* add on_kill override to databricks workflow operator * on_kill equivalent for DatabricksSqlOperator * add tests for create_timeout_thread * add note for on_kill in DatabricksCopyIntoOperator * chore: static checks * remove changes for databricks_sql.py for PR isolated to databricks_workflows.py --------- Co-authored-by: Lorin <lorin@databricks.com>

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed. DatabricksTaskBaseOperator — used by DatabricksTaskOperator and DatabricksNotebookOperator — is missing the same implementation, so Databricks jobs continue running after the Airflow task is killed, orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. Uses self.databricks_run_id which is initialised to None in __init__ and set by _launch_job() the moment the run is submitted.

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, and workflow-member regression (parent run_id=1, child run_id=999, asserts cancel_run(999)).

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

…ll (#69442) DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in #42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

…ll (apache#69442) DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

R7L208 · 2024-09-09T16:25:10Z

The Databricks Provider did not implement on_kill to cancel tasks generated by _CreateDatabricksWorkflowOperator. This led to data quality issues, where Airflow would report a cancellation due to timeout; however, the corresponding workflow task would continue to run on Databricks.

This PR implements on_kill for _CreateDatabricksWorkflowOperator.

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

boring-cyborg · 2024-09-09T16:25:15Z

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:

Pay attention to the quality of your code (ruff, mypy and type annotations). Our pre-commits will help you with that.
In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
Be sure to read the Airflow Coding style.
Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack

R7L208 · 2024-10-01T23:57:42Z

hey @pankajkoti - Any ETA on when you'd be able to review this PR? 🙏

pankajkoti

hi @R7L208 Apologies, I missed reviewing it so far.

I am happy to review the changes for databricks_workflow.py but not so comfortable yet reviewing the changes in databricks_sql.py (operators & hooks) as I do not have enough expertise on the changes made there. Would you please to separate out the two changes(move out databricks_sql.py related changes to another PR) and then we can invite some expert to review the other PR?

R7L208 · 2024-10-02T14:43:45Z

Opened #42668

@pankajkoti @eladkal - can you please assign someone to review?

pankajkoti · 2024-10-02T15:09:30Z

@R7L208 Can you please update the PR description based on the altered scope of this PR?

R7L208 · 2024-10-02T17:36:03Z

@pankajkoti - it's been updated

…rkflows.py

pankajkoti · 2024-10-02T17:45:26Z

@pankajkoti - it's been updated

@R7L208 it's not. It mentions about DatabricksSqlHook queries ("SQL queries submitted by DatabricksSqlHook") & using threading ("uses threading to cancel SQL queries submitted by DatabricksSqlHook.run()") which we no longer have in this PR. Could you please re-read the description and update to what's limited to the scope of this PR.

Also, _CreateDatabricksWorkflowOperator does not rely on DatabricksSqlHook, but leverages DatabricksHook just in case you missed checking that :)

R7L208 · 2024-10-02T17:50:58Z

@pankajkoti - Apologies! I read PR title instead of PR description 🤦

PR description is now updated

boring-cyborg · 2024-10-03T12:36:32Z

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.

* add on_kill override to databricks workflow operator * on_kill equivalent for DatabricksSqlOperator * add tests for create_timeout_thread * add note for on_kill in DatabricksCopyIntoOperator * chore: static checks * remove changes for databricks_sql.py for PR isolated to databricks_workflows.py --------- Co-authored-by: Lorin <lorin@databricks.com>

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed. DatabricksTaskBaseOperator — used by DatabricksTaskOperator and DatabricksNotebookOperator — is missing the same implementation, so Databricks jobs continue running after the Airflow task is killed, orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. Uses self.databricks_run_id which is initialised to None in __init__ and set by _launch_job() the moment the run is submitted.

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, and workflow-member regression (parent run_id=1, child run_id=999, asserts cancel_run(999)).

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

…ll (#69442) DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in #42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

…ll (apache#69442) DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

R7L208 · 2024-09-09T16:25:10Z

The Databricks Provider did not implement on_kill to cancel tasks generated by _CreateDatabricksWorkflowOperator. This led to data quality issues, where Airflow would report a cancellation due to timeout; however, the corresponding workflow task would continue to run on Databricks.

This PR implements on_kill for _CreateDatabricksWorkflowOperator.

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

boring-cyborg · 2024-09-09T16:25:15Z

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:

Pay attention to the quality of your code (ruff, mypy and type annotations). Our pre-commits will help you with that.
In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
Be sure to read the Airflow Coding style.
Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack

R7L208 · 2024-10-01T23:57:42Z

hey @pankajkoti - Any ETA on when you'd be able to review this PR? 🙏

pankajkoti

hi @R7L208 Apologies, I missed reviewing it so far.

I am happy to review the changes for databricks_workflow.py but not so comfortable yet reviewing the changes in databricks_sql.py (operators & hooks) as I do not have enough expertise on the changes made there. Would you please to separate out the two changes(move out databricks_sql.py related changes to another PR) and then we can invite some expert to review the other PR?

R7L208 · 2024-10-02T14:43:45Z

Opened #42668

@pankajkoti @eladkal - can you please assign someone to review?

pankajkoti · 2024-10-02T15:09:30Z

@R7L208 Can you please update the PR description based on the altered scope of this PR?

R7L208 · 2024-10-02T17:36:03Z

@pankajkoti - it's been updated

…rkflows.py

pankajkoti · 2024-10-02T17:45:26Z

@pankajkoti - it's been updated

@R7L208 it's not. It mentions about DatabricksSqlHook queries ("SQL queries submitted by DatabricksSqlHook") & using threading ("uses threading to cancel SQL queries submitted by DatabricksSqlHook.run()") which we no longer have in this PR. Could you please re-read the description and update to what's limited to the scope of this PR.

Also, _CreateDatabricksWorkflowOperator does not rely on DatabricksSqlHook, but leverages DatabricksHook just in case you missed checking that :)

R7L208 · 2024-10-02T17:50:58Z

@pankajkoti - Apologies! I read PR title instead of PR description 🤦

PR description is now updated

boring-cyborg · 2024-10-03T12:36:32Z

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.

* add on_kill override to databricks workflow operator * on_kill equivalent for DatabricksSqlOperator * add tests for create_timeout_thread * add note for on_kill in DatabricksCopyIntoOperator * chore: static checks * remove changes for databricks_sql.py for PR isolated to databricks_workflows.py --------- Co-authored-by: Lorin <lorin@databricks.com>

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed. DatabricksTaskBaseOperator — used by DatabricksTaskOperator and DatabricksNotebookOperator — is missing the same implementation, so Databricks jobs continue running after the Airflow task is killed, orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. Uses self.databricks_run_id which is initialised to None in __init__ and set by _launch_job() the moment the run is submitted.

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, and workflow-member regression (parent run_id=1, child run_id=999, asserts cancel_run(999)).

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

…ll (#69442) DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in #42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

…ll (apache#69442) DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

R7L208 · 2024-09-09T16:25:10Z

The Databricks Provider did not implement on_kill to cancel tasks generated by _CreateDatabricksWorkflowOperator. This led to data quality issues, where Airflow would report a cancellation due to timeout; however, the corresponding workflow task would continue to run on Databricks.

This PR implements on_kill for _CreateDatabricksWorkflowOperator.

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

boring-cyborg · 2024-09-09T16:25:15Z

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:

Pay attention to the quality of your code (ruff, mypy and type annotations). Our pre-commits will help you with that.
In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
Be sure to read the Airflow Coding style.
Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack

R7L208 · 2024-10-01T23:57:42Z

hey @pankajkoti - Any ETA on when you'd be able to review this PR? 🙏

pankajkoti

hi @R7L208 Apologies, I missed reviewing it so far.

I am happy to review the changes for databricks_workflow.py but not so comfortable yet reviewing the changes in databricks_sql.py (operators & hooks) as I do not have enough expertise on the changes made there. Would you please to separate out the two changes(move out databricks_sql.py related changes to another PR) and then we can invite some expert to review the other PR?

R7L208 · 2024-10-02T14:43:45Z

Opened #42668

@pankajkoti @eladkal - can you please assign someone to review?

pankajkoti · 2024-10-02T15:09:30Z

@R7L208 Can you please update the PR description based on the altered scope of this PR?

R7L208 · 2024-10-02T17:36:03Z

@pankajkoti - it's been updated

…rkflows.py

pankajkoti · 2024-10-02T17:45:26Z

@pankajkoti - it's been updated

@R7L208 it's not. It mentions about DatabricksSqlHook queries ("SQL queries submitted by DatabricksSqlHook") & using threading ("uses threading to cancel SQL queries submitted by DatabricksSqlHook.run()") which we no longer have in this PR. Could you please re-read the description and update to what's limited to the scope of this PR.

Also, _CreateDatabricksWorkflowOperator does not rely on DatabricksSqlHook, but leverages DatabricksHook just in case you missed checking that :)

R7L208 · 2024-10-02T17:50:58Z

@pankajkoti - Apologies! I read PR title instead of PR description 🤦

PR description is now updated

boring-cyborg · 2024-10-03T12:36:32Z

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.

* add on_kill override to databricks workflow operator * on_kill equivalent for DatabricksSqlOperator * add tests for create_timeout_thread * add note for on_kill in DatabricksCopyIntoOperator * chore: static checks * remove changes for databricks_sql.py for PR isolated to databricks_workflows.py --------- Co-authored-by: Lorin <lorin@databricks.com>

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed. DatabricksTaskBaseOperator — used by DatabricksTaskOperator and DatabricksNotebookOperator — is missing the same implementation, so Databricks jobs continue running after the Airflow task is killed, orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. Uses self.databricks_run_id which is initialised to None in __init__ and set by _launch_job() the moment the run is submitted.

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, and workflow-member regression (parent run_id=1, child run_id=999, asserts cancel_run(999)).

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

…ll (#69442) DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in #42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

…ll (apache#69442) DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

R7L208 · 2024-09-09T16:25:10Z

The Databricks Provider did not implement on_kill to cancel tasks generated by _CreateDatabricksWorkflowOperator. This led to data quality issues, where Airflow would report a cancellation due to timeout; however, the corresponding workflow task would continue to run on Databricks.

This PR implements on_kill for _CreateDatabricksWorkflowOperator.

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

boring-cyborg · 2024-09-09T16:25:15Z

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:

Pay attention to the quality of your code (ruff, mypy and type annotations). Our pre-commits will help you with that.
In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
Be sure to read the Airflow Coding style.
Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack

R7L208 · 2024-10-01T23:57:42Z

hey @pankajkoti - Any ETA on when you'd be able to review this PR? 🙏

pankajkoti

hi @R7L208 Apologies, I missed reviewing it so far.

I am happy to review the changes for databricks_workflow.py but not so comfortable yet reviewing the changes in databricks_sql.py (operators & hooks) as I do not have enough expertise on the changes made there. Would you please to separate out the two changes(move out databricks_sql.py related changes to another PR) and then we can invite some expert to review the other PR?

R7L208 · 2024-10-02T14:43:45Z

Opened #42668

@pankajkoti @eladkal - can you please assign someone to review?

pankajkoti · 2024-10-02T15:09:30Z

@R7L208 Can you please update the PR description based on the altered scope of this PR?

R7L208 · 2024-10-02T17:36:03Z

@pankajkoti - it's been updated

…rkflows.py

pankajkoti · 2024-10-02T17:45:26Z

@pankajkoti - it's been updated

@R7L208 it's not. It mentions about DatabricksSqlHook queries ("SQL queries submitted by DatabricksSqlHook") & using threading ("uses threading to cancel SQL queries submitted by DatabricksSqlHook.run()") which we no longer have in this PR. Could you please re-read the description and update to what's limited to the scope of this PR.

Also, _CreateDatabricksWorkflowOperator does not rely on DatabricksSqlHook, but leverages DatabricksHook just in case you missed checking that :)

R7L208 · 2024-10-02T17:50:58Z

@pankajkoti - Apologies! I read PR title instead of PR description 🤦

PR description is now updated

boring-cyborg · 2024-10-03T12:36:32Z

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.

* add on_kill override to databricks workflow operator * on_kill equivalent for DatabricksSqlOperator * add tests for create_timeout_thread * add note for on_kill in DatabricksCopyIntoOperator * chore: static checks * remove changes for databricks_sql.py for PR isolated to databricks_workflows.py --------- Co-authored-by: Lorin <lorin@databricks.com>

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed. DatabricksTaskBaseOperator — used by DatabricksTaskOperator and DatabricksNotebookOperator — is missing the same implementation, so Databricks jobs continue running after the Airflow task is killed, orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. Uses self.databricks_run_id which is initialised to None in __init__ and set by _launch_job() the moment the run is submitted.

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, and workflow-member regression (parent run_id=1, child run_id=999, asserts cancel_run(999)).

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

…ll (#69442) DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in #42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

…ll (apache#69442) DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

R7L208 · 2024-09-09T16:25:10Z

The Databricks Provider did not implement on_kill to cancel tasks generated by _CreateDatabricksWorkflowOperator. This led to data quality issues, where Airflow would report a cancellation due to timeout; however, the corresponding workflow task would continue to run on Databricks.

This PR implements on_kill for _CreateDatabricksWorkflowOperator.

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

boring-cyborg · 2024-09-09T16:25:15Z

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:

Pay attention to the quality of your code (ruff, mypy and type annotations). Our pre-commits will help you with that.
In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
Be sure to read the Airflow Coding style.
Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack

R7L208 · 2024-10-01T23:57:42Z

hey @pankajkoti - Any ETA on when you'd be able to review this PR? 🙏

pankajkoti

hi @R7L208 Apologies, I missed reviewing it so far.

I am happy to review the changes for databricks_workflow.py but not so comfortable yet reviewing the changes in databricks_sql.py (operators & hooks) as I do not have enough expertise on the changes made there. Would you please to separate out the two changes(move out databricks_sql.py related changes to another PR) and then we can invite some expert to review the other PR?

R7L208 · 2024-10-02T14:43:45Z

Opened #42668

@pankajkoti @eladkal - can you please assign someone to review?

pankajkoti · 2024-10-02T15:09:30Z

@R7L208 Can you please update the PR description based on the altered scope of this PR?

R7L208 · 2024-10-02T17:36:03Z

@pankajkoti - it's been updated

…rkflows.py

pankajkoti · 2024-10-02T17:45:26Z

@pankajkoti - it's been updated

@R7L208 it's not. It mentions about DatabricksSqlHook queries ("SQL queries submitted by DatabricksSqlHook") & using threading ("uses threading to cancel SQL queries submitted by DatabricksSqlHook.run()") which we no longer have in this PR. Could you please re-read the description and update to what's limited to the scope of this PR.

Also, _CreateDatabricksWorkflowOperator does not rely on DatabricksSqlHook, but leverages DatabricksHook just in case you missed checking that :)

R7L208 · 2024-10-02T17:50:58Z

@pankajkoti - Apologies! I read PR title instead of PR description 🤦

PR description is now updated

boring-cyborg · 2024-10-03T12:36:32Z

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.

* add on_kill override to databricks workflow operator * on_kill equivalent for DatabricksSqlOperator * add tests for create_timeout_thread * add note for on_kill in DatabricksCopyIntoOperator * chore: static checks * remove changes for databricks_sql.py for PR isolated to databricks_workflows.py --------- Co-authored-by: Lorin <lorin@databricks.com>

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed. DatabricksTaskBaseOperator — used by DatabricksTaskOperator and DatabricksNotebookOperator — is missing the same implementation, so Databricks jobs continue running after the Airflow task is killed, orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. Uses self.databricks_run_id which is initialised to None in __init__ and set by _launch_job() the moment the run is submitted.

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, and workflow-member regression (parent run_id=1, child run_id=999, asserts cancel_run(999)).

DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

…ll (#69442) DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in #42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

…ll (apache#69442) DatabricksSubmitRunOperator and DatabricksRunNowOperator both implement on_kill() to cancel the Databricks run when an Airflow task is killed (SIGTERM or execution_timeout). DatabricksTaskBaseOperator — the base for DatabricksTaskOperator and DatabricksNotebookOperator — was missing the same implementation, leaving Databricks jobs running after the Airflow task was killed and orphaning compute resources. DatabricksWorkflowTaskGroup received on_kill() in apache#42115; this PR closes the remaining gap for standalone task operators. For workflow members self.databricks_run_id is the shared parent run ID; cancelling it would stop all sibling tasks. on_kill() therefore calls _get_current_databricks_task()["run_id"] to target only the current task's own child run, mirroring monitor_databricks_job. Standalone operators continue to cancel via self.databricks_run_id directly. If resolving the child run_id fails (API error, task_key mismatch), on_kill logs the exception and returns without cancelling anything — falling back to the parent run_id would stop sibling tasks, defeating the purpose. Unit tests cover: cancel called for standalone operator, no-op when databricks_run_id is None, workflow-member cancels child run (parent=1, child=999, asserts cancel_run(999)), and workflow-member where _get_current_databricks_task raises asserts cancel_run not called.

boring-cyborgBot added area:providers provider:databricks labels Sep 9, 2024

R7L208 force-pushed the lorin/databricks-operators-on_kill-override branch 5 times, most recently from 70f8c96 to c3f1e9bCompare September 13, 2024 15:48

R7L208 force-pushed the lorin/databricks-operators-on_kill-override branch from c3f1e9b to b80e162Compare September 18, 2024 19:31

eladkal requested a review from pankajkoti September 19, 2024 04:37

R7L208 force-pushed the lorin/databricks-operators-on_kill-override branch 2 times, most recently from 8a86b3b to ac988bbCompare September 27, 2024 16:38

R7L208 force-pushed the lorin/databricks-operators-on_kill-override branch from ac988bb to 532feb3Compare October 1, 2024 23:55

pankajkoti reviewed Oct 2, 2024
View reviewed changes

R7L208 force-pushed the lorin/databricks-operators-on_kill-override branch from 532feb3 to 10ceb61Compare October 2, 2024 14:28

R7L208 changed the title ~~Add on_kill or equivalent to Databricks Operators/Hooks to cancel timed out queries~~Add on_kill to Databricks Workflow Operator to cancel timed out queriesOct 2, 2024

R7L208 force-pushed the lorin/databricks-operators-on_kill-override branch from d45c9d9 to bf5237aCompare October 2, 2024 14:45

pankajkoti requested a review from Lee-W October 2, 2024 15:08

pankajkoti approved these changes Oct 2, 2024
View reviewed changes

R7L208 changed the title ~~Add on_kill to Databricks Workflow Operator to cancel timed out queries~~Add on_kill to Databricks Workflow OperatorOct 2, 2024

Lorin added 6 commits October 2, 2024 11:36

add on_kill override to databricks workflow operator
444651e

on_kill equivalent for DatabricksSqlOperator
d813864

add tests for create_timeout_thread
532c7ee

add note for on_kill in DatabricksCopyIntoOperator
1cba0d6

chore: static checks
5696ee9

remove changes for databricks_sql.py for PR isolated to databricks_wo…
bf0555b
…rkflows.py

R7L208 force-pushed the lorin/databricks-operators-on_kill-override branch from bf5237a to bf0555bCompare October 2, 2024 17:36

Lee-W approved these changes Oct 3, 2024
View reviewed changes

Lee-W merged commit 5d51bee into apache:mainOct 3, 2024

eladkal mentioned this pull request Oct 10, 2024
Status of testing Providers that were prepared on October 10, 2024 #42882
Closed
96 tasks

eladkal mentioned this pull request Nov 14, 2024
Status of testing Providers that were prepared on November 14, 2024 #44041
Closed

victorymakes mentioned this pull request Jul 6, 2026
Add on_kill() to DatabricksTaskBaseOperator to cancel runs on task kill #69442
Merged
1 task