You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "romsharon98 (via GitHub)" <gi...@apache.org> on 2023/12/22 20:30:41 UTC
[PR] draft: Add on skipped callback [airflow]
romsharon98 opened a new pull request, #36374:
URL: https://github.com/apache/airflow/pull/36374
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
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](https://github.com/apache/airflow/tree/main/newsfragments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `on_skipped_callback` in to `BaseOperator` [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #36374:
URL: https://github.com/apache/airflow/pull/36374#discussion_r1451800447
##########
airflow/models/taskinstance.py:
##########
@@ -2361,6 +2361,8 @@ def _run_raw_task(
self.log.info(e)
if not test_mode:
self.refresh_from_db(lock_for_update=True, session=session)
+ _run_finished_callback(callbacks=self.task.on_skipped_callback, context=context)
+ session.commit()
Review Comment:
I see a lot of other approvals. Resolving this one - it always can come back here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `on_skipped_callback` in to `BaseOperator` [airflow]
Posted by "romsharon98 (via GitHub)" <gi...@apache.org>.
romsharon98 commented on PR #36374:
URL: https://github.com/apache/airflow/pull/36374#issuecomment-1868309042
> The trigger rule dep could change the TI state to skipped without raising this exception, I wonder if we should handle that case too 🤔
>
> https://github.com/apache/airflow/blob/95bf5dd620ec996dda834ba13048a77314d6d915/airflow/ti_deps/deps/trigger_rule_dep.py#L285
>
> I'm not sure if we handle that for other callbacks
As @eladkal commented, I think it should take care by another PR that call the `on_skipped_callback` everywhere the task is skipped and not only if this exception is thrown.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `on_skipped_callback` in to `BaseOperator` [airflow]
Posted by "dirrao (via GitHub)" <gi...@apache.org>.
dirrao commented on PR #36374:
URL: https://github.com/apache/airflow/pull/36374#issuecomment-1868228237
> The trigger rule dep could change the TI state to skipped without raising this exception, I wonder if we should handle that case too 🤔
>
> https://github.com/apache/airflow/blob/95bf5dd620ec996dda834ba13048a77314d6d915/airflow/ti_deps/deps/trigger_rule_dep.py#L285
>
> I'm not sure if we handle that for other callbacks
I have seen a scenario where the skipped state is set when branching doesn't follow some of the downstream tasks. Looks like that's not covered here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `on_skipped_callback` in to `BaseOperator` [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #36374:
URL: https://github.com/apache/airflow/pull/36374#discussion_r1436430707
##########
airflow/models/taskinstance.py:
##########
@@ -2361,6 +2361,8 @@ def _run_raw_task(
self.log.info(e)
if not test_mode:
self.refresh_from_db(lock_for_update=True, session=session)
+ _run_finished_callback(callbacks=self.task.on_skipped_callback, context=context)
+ session.commit()
Review Comment:
Hmm. In other exception handling code we already commit even if test_mode. I wonder if it is a bug (and which)?
```
except AirflowException as e:
if not test_mode:
self.refresh_from_db(lock_for_update=True, session=session)
# for case when task is marked as success/failed externally
# or dagrun timed out and task is marked as skipped
# current behavior doesn't hit the callbacks
if self.state in State.finished:
self.clear_next_method_args()
session.merge(self)
session.commit()
return None
else:
self.handle_failure(e, test_mode, context, session=session)
session.commit()
raise
except (Exception, KeyboardInterrupt) as e:
self.handle_failure(e, test_mode, context, session=session)
session.commit()
raise
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `on_skipped_callback` in to `BaseOperator` [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #36374:
URL: https://github.com/apache/airflow/pull/36374#discussion_r1438943686
##########
airflow/models/baseoperator.py:
##########
@@ -597,6 +599,8 @@ class derived from this one results in the creation of a task object,
that it is executed when retries occur.
:param on_success_callback: much like the ``on_failure_callback`` except
that it is executed when the task succeeds.
+ :param on_skipped_callback: much like the ``on_failure_callback`` except
+ that it is executed when skipped occur; this callback will be called only if AirflowSkipException get raised.
Review Comment:
I'd favor to make it explicit. Many people/starters might not know about this exception.
```suggestion
:param on_skipped_callback: much like the ``on_failure_callback`` except
that it is executed when skipped occur; this callback will be called only if AirflowSkipException get raised.
Explicitly it is NOT called if a task is not started to be executed because of a preceding branching
decision in the DAG or a trigger rule which causes execution to skip so that the task execution
is never scheduled.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `on_skipped_callback` in to `BaseOperator` [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk merged PR #36374:
URL: https://github.com/apache/airflow/pull/36374
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `on_skipped_callback` in to `BaseOperator` [airflow]
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on PR #36374:
URL: https://github.com/apache/airflow/pull/36374#issuecomment-1868617680
The mentioned issue mentions #30264 as a potential use case, but this PR will not close it, because the running tasks are marked as skipped without raising any skip exception:
https://github.com/apache/airflow/blob/c7e13061bd472962ed571cb0d33784900217660a/airflow/jobs/scheduler_job_runner.py#L1449-L1464
So IMHO, if you want to split implementing the callback on different PRs, you need to update the description by replacing:
```md
closes: #35936
```
by
```md
related: #35936
```
of course, with clarifying that in the doc as @eladkal suggested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `on_skipped_callback` in to `BaseOperator` [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #36374:
URL: https://github.com/apache/airflow/pull/36374#discussion_r1435960507
##########
airflow/models/taskinstance.py:
##########
@@ -2361,6 +2361,8 @@ def _run_raw_task(
self.log.info(e)
if not test_mode:
self.refresh_from_db(lock_for_update=True, session=session)
+ _run_finished_callback(callbacks=self.task.on_skipped_callback, context=context)
+ session.commit()
Review Comment:
This should not be run in test mode, similar to other callbacks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `on_skipped_callback` in to `BaseOperator` [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #36374:
URL: https://github.com/apache/airflow/pull/36374#discussion_r1438943838
##########
docs/apache-airflow/administration-and-deployment/logging-monitoring/callbacks.rst:
##########
@@ -49,6 +49,7 @@ Name Description
``sla_miss_callback`` Invoked when a task misses its defined :ref:`SLA <concepts:slas>`
``on_retry_callback`` Invoked when the task is :ref:`up for retry <concepts:task-instances>`
``on_execute_callback`` Invoked right before the task begins executing.
+``on_skipped_callback`` Invoked when the task is :ref:`running <concepts:task-instances>` and AirflowSkipException raised.
Review Comment:
+1 here:
```suggestion
``on_skipped_callback`` Invoked when the task is :ref:`running <concepts:task-instances>` and AirflowSkipException raised.
Explicitly it is NOT called if a task is not started to be executed because of a preceding branching
decision in the DAG or a trigger rule which causes execution to skip so that the task execution
is never scheduled.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `on_skipped_callback` in to `BaseOperator` [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #36374:
URL: https://github.com/apache/airflow/pull/36374#issuecomment-1869494704
> If you are OK with it I think we can keep it as close and open a new issue that aim to solve the 2nd use case. Its just easier to track when issue has single well defined task.
I just updated the description to "related: #35936" instead of `closes:` . That's enough to make the issue related but not close the original issue when merged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `on_skipped_callback` in to `BaseOperator` [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on PR #36374:
URL: https://github.com/apache/airflow/pull/36374#issuecomment-1868301645
> I have seen a scenario where the skipped state is set when branching doesn't follow some of the downstream tasks. Looks like that's not covered.
I'm not sure if we can/should handle it. The goal here is to handle the case when the task is executed and then status changed to skipped due to raise of skip exception. It is not meant to be used with cascading status to downstream tasks (though maybe it's something we should think about). However for the scope of this PR I think we should clarify in docs about this unique case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `on_skipped_callback` in to `BaseOperator` [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on PR #36374:
URL: https://github.com/apache/airflow/pull/36374#issuecomment-1868770241
> The mentioned issue mentions #30264 as a potential use case, but this PR will not close it
That was my edit to the description.
If you are OK with it I think we can keep it as close and open a new issue that aim to solve the 2nd use case. Its just easier to track when issue has single well defined task.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `on_skipped_callback` in to `BaseOperator` [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #36374:
URL: https://github.com/apache/airflow/pull/36374#discussion_r1450339055
##########
airflow/models/taskinstance.py:
##########
@@ -2361,6 +2361,8 @@ def _run_raw_task(
self.log.info(e)
if not test_mode:
self.refresh_from_db(lock_for_update=True, session=session)
+ _run_finished_callback(callbacks=self.task.on_skipped_callback, context=context)
+ session.commit()
Review Comment:
@uranusjr - any coments for that one, or can we just resolve it ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org