Fix double cancellation for airflow 3.3+ for EmrServerlessStartJobTrigger

[amoghrajesh]

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

related to #65740 which added on_kill() to EmrServerlessStartJobTrigger to cancel the EMR Serverless job when a user kills the task. However, the existing CancelledError handler in run() was left untouched which means that on Airflow 3.3 onwards (or main branch for that matter), both paths would fire for the same user kill, sending cancel_job_run twice.

This PR guards the CancelledError handler with the __airflow_user_action__ sentinel that Airflow 3.3 sets when cancelling a trigger due to a user action. On 3.3+, on_kill() handles the cancellation and the CancelledError handler skips it. On older Airflow where the sentinel doesn't exist, the handler runs as before — so backward compatibility is preserved.

---

• Read the 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) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[vincbeck]

Good catch! Thanks

[amoghrajesh]

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

related to #65740 which added on_kill() to EmrServerlessStartJobTrigger to cancel the EMR Serverless job when a user kills the task. However, the existing CancelledError handler in run() was left untouched which means that on Airflow 3.3 onwards (or main branch for that matter), both paths would fire for the same user kill, sending cancel_job_run twice.

This PR guards the CancelledError handler with the __airflow_user_action__ sentinel that Airflow 3.3 sets when cancelling a trigger due to a user action. On 3.3+, on_kill() handles the cancellation and the CancelledError handler skips it. On older Airflow where the sentinel doesn't exist, the handler runs as before — so backward compatibility is preserved.

---

• Read the 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) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[vincbeck]

Good catch! Thanks

[amoghrajesh]

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

related to #65740 which added on_kill() to EmrServerlessStartJobTrigger to cancel the EMR Serverless job when a user kills the task. However, the existing CancelledError handler in run() was left untouched which means that on Airflow 3.3 onwards (or main branch for that matter), both paths would fire for the same user kill, sending cancel_job_run twice.

This PR guards the CancelledError handler with the __airflow_user_action__ sentinel that Airflow 3.3 sets when cancelling a trigger due to a user action. On 3.3+, on_kill() handles the cancellation and the CancelledError handler skips it. On older Airflow where the sentinel doesn't exist, the handler runs as before — so backward compatibility is preserved.

---

• Read the 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) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[vincbeck]

Good catch! Thanks

[amoghrajesh]

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

related to #65740 which added on_kill() to EmrServerlessStartJobTrigger to cancel the EMR Serverless job when a user kills the task. However, the existing CancelledError handler in run() was left untouched which means that on Airflow 3.3 onwards (or main branch for that matter), both paths would fire for the same user kill, sending cancel_job_run twice.

This PR guards the CancelledError handler with the __airflow_user_action__ sentinel that Airflow 3.3 sets when cancelling a trigger due to a user action. On 3.3+, on_kill() handles the cancellation and the CancelledError handler skips it. On older Airflow where the sentinel doesn't exist, the handler runs as before — so backward compatibility is preserved.

---

• Read the 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) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[vincbeck]

Good catch! Thanks

[amoghrajesh]

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

related to #65740 which added on_kill() to EmrServerlessStartJobTrigger to cancel the EMR Serverless job when a user kills the task. However, the existing CancelledError handler in run() was left untouched which means that on Airflow 3.3 onwards (or main branch for that matter), both paths would fire for the same user kill, sending cancel_job_run twice.

This PR guards the CancelledError handler with the __airflow_user_action__ sentinel that Airflow 3.3 sets when cancelling a trigger due to a user action. On 3.3+, on_kill() handles the cancellation and the CancelledError handler skips it. On older Airflow where the sentinel doesn't exist, the handler runs as before — so backward compatibility is preserved.

---

• Read the 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) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[vincbeck]

Good catch! Thanks

[amoghrajesh]

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

related to #65740 which added on_kill() to EmrServerlessStartJobTrigger to cancel the EMR Serverless job when a user kills the task. However, the existing CancelledError handler in run() was left untouched which means that on Airflow 3.3 onwards (or main branch for that matter), both paths would fire for the same user kill, sending cancel_job_run twice.

This PR guards the CancelledError handler with the __airflow_user_action__ sentinel that Airflow 3.3 sets when cancelling a trigger due to a user action. On 3.3+, on_kill() handles the cancellation and the CancelledError handler skips it. On older Airflow where the sentinel doesn't exist, the handler runs as before — so backward compatibility is preserved.

---

• Read the 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) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[vincbeck]

Good catch! Thanks

[amoghrajesh]

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

related to #65740 which added on_kill() to EmrServerlessStartJobTrigger to cancel the EMR Serverless job when a user kills the task. However, the existing CancelledError handler in run() was left untouched which means that on Airflow 3.3 onwards (or main branch for that matter), both paths would fire for the same user kill, sending cancel_job_run twice.

This PR guards the CancelledError handler with the __airflow_user_action__ sentinel that Airflow 3.3 sets when cancelling a trigger due to a user action. On 3.3+, on_kill() handles the cancellation and the CancelledError handler skips it. On older Airflow where the sentinel doesn't exist, the handler runs as before — so backward compatibility is preserved.

---

• Read the 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) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[vincbeck]

Good catch! Thanks

[amoghrajesh]

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

related to #65740 which added on_kill() to EmrServerlessStartJobTrigger to cancel the EMR Serverless job when a user kills the task. However, the existing CancelledError handler in run() was left untouched which means that on Airflow 3.3 onwards (or main branch for that matter), both paths would fire for the same user kill, sending cancel_job_run twice.

This PR guards the CancelledError handler with the __airflow_user_action__ sentinel that Airflow 3.3 sets when cancelling a trigger due to a user action. On 3.3+, on_kill() handles the cancellation and the CancelledError handler skips it. On older Airflow where the sentinel doesn't exist, the handler runs as before — so backward compatibility is preserved.

---

• Read the 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) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[vincbeck]

Good catch! Thanks