Add on_skipped_callback in to BaseOperator

[romsharon98]

related: #35936

---

^ 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.

[hussein-awala]

The trigger rule dep could change the TI state to skipped without raising this exception, I wonder if we should handle that case too 🤔

airflow/airflow/ti_deps/deps/trigger_rule_dep.py

Line 285 in 95bf5dd

changed = ti.set_state(new_state, session)

I'm not sure if we handle that for other callbacks

[dirrao]

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.

[eladkal]

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.

[romsharon98]

The trigger rule dep could change the TI state to skipped without raising this exception, I wonder if we should handle that case too 🤔

airflow/airflow/ti_deps/deps/trigger_rule_dep.py

Line 285 in 95bf5dd

changed = ti.set_state(new_state, session)

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.

[hussein-awala]

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:

airflow/airflow/jobs/scheduler_job_runner.py

Lines 1449 to 1464 in c7e1306

	if (
	dag_run.start_date
	and dag.dagrun_timeout
	and dag_run.start_date < timezone.utcnow() - dag.dagrun_timeout
	):
	dag_run.set_state(DagRunState.FAILED)
	unfinished_task_instances = session.scalars(
	select(TI)
	.where(TI.dag_id == dag_run.dag_id)
	.where(TI.run_id == dag_run.run_id)
	.where(TI.state.in_(State.unfinished))
	)
	for task_instance in unfinished_task_instances:
	task_instance.state = TaskInstanceState.SKIPPED
	session.merge(task_instance)
	session.flush()

So IMHO, if you want to split implementing the callback on different PRs, you need to update the description by replacing:

closes: #35936

by

related: #35936

of course, with clarifying that in the doc as @eladkal suggested.

[eladkal]

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.

[potiuk]

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.