You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "kaxil (via GitHub)" <gi...@apache.org> on 2025/10/16 13:22:03 UTC
[PR] Improve DAG and task missing error handling in callbacks [airflow]
kaxil opened a new pull request, #56725:
URL: https://github.com/apache/airflow/pull/56725
When callbacks are executed but the DAG or task is missing from the `DagBag`, users now receive clear error messages explaining what went wrong instead of cryptic `KeyError` exceptions. This typically occurs due to race conditions where DAGs are removed or fail to parse between scheduling and callback execution.
Fixes #56701
<!--
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-docs/05_pull_requests.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 [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/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] Improve DAG and task missing error handling in callbacks [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #56725:
URL: https://github.com/apache/airflow/pull/56725#discussion_r2439794312
##########
airflow-core/src/airflow/dag_processing/processor.py:
##########
@@ -234,6 +238,39 @@ def _serialize_dags(
return serialized_dags, serialization_import_errors
+def _get_dag_with_task(
+ dagbag: DagBag, dag_id: str, task_id: str | None = None
+) -> tuple[DAG, BaseOperator | MappedOperator | None]:
+ """
+ Retrieve a DAG and optionally a task from the DagBag.
+
+ :param dagbag: DagBag to retrieve from
+ :param dag_id: DAG ID to retrieve
+ :param task_id: Optional task ID to retrieve from the DAG
+ :return: tuple of (dag, task) where task is None if not requested
+ :raises ValueError: If DAG or task is not found
+ """
+ if dag_id not in dagbag.dags:
+ raise ValueError(
+ f"DAG '{dag_id}' not found in DagBag. "
+ f"This typically indicates a race condition where the DAG was removed or failed to parse."
+ )
+
+ dag = dagbag.dags[dag_id]
+
+ if task_id is not None:
+ try:
+ task = dag.get_task(task_id)
+ return dag, task
+ except TaskNotFound:
+ raise ValueError(
+ f"Task '{task_id}' not found in DAG '{dag_id}'. "
+ f"This typically indicates a race condition where the task was removed or the DAG structure changed."
Review Comment:
Eh .. 🤷 This is mainly for the logs -- as long as you see the message clearly defined -- what's the necessity of `DagNotFoundException`.
Feel free to change it though -- I don't feel too strongly about it. I do not expect anyone to catch those errors in dag processor.
--
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] Improve DAG and task missing error handling in callbacks [airflow]
Posted by "rawwar (via GitHub)" <gi...@apache.org>.
rawwar commented on code in PR #56725:
URL: https://github.com/apache/airflow/pull/56725#discussion_r2438827396
##########
airflow-core/src/airflow/dag_processing/processor.py:
##########
@@ -234,6 +238,39 @@ def _serialize_dags(
return serialized_dags, serialization_import_errors
+def _get_dag_with_task(
+ dagbag: DagBag, dag_id: str, task_id: str | None = None
+) -> tuple[DAG, BaseOperator | MappedOperator | None]:
+ """
+ Retrieve a DAG and optionally a task from the DagBag.
+
+ :param dagbag: DagBag to retrieve from
+ :param dag_id: DAG ID to retrieve
+ :param task_id: Optional task ID to retrieve from the DAG
+ :return: tuple of (dag, task) where task is None if not requested
+ :raises ValueError: If DAG or task is not found
+ """
+ if dag_id not in dagbag.dags:
+ raise ValueError(
+ f"DAG '{dag_id}' not found in DagBag. "
+ f"This typically indicates a race condition where the DAG was removed or failed to parse."
+ )
+
+ dag = dagbag.dags[dag_id]
+
+ if task_id is not None:
+ try:
+ task = dag.get_task(task_id)
+ return dag, task
+ except TaskNotFound:
+ raise ValueError(
+ f"Task '{task_id}' not found in DAG '{dag_id}'. "
+ f"This typically indicates a race condition where the task was removed or the DAG structure changed."
Review Comment:
at least the error messages that User see's, they should be consistent and we can maintain such consistency by using such user defined exceptions.
--
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] Improve DAG and task missing error handling in callbacks [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #56725:
URL: https://github.com/apache/airflow/pull/56725#discussion_r2438009180
##########
airflow-core/src/airflow/dag_processing/processor.py:
##########
@@ -234,6 +238,39 @@ def _serialize_dags(
return serialized_dags, serialization_import_errors
+def _get_dag_with_task(
+ dagbag: DagBag, dag_id: str, task_id: str | None = None
+) -> tuple[DAG, BaseOperator | MappedOperator | None]:
+ """
+ Retrieve a DAG and optionally a task from the DagBag.
+
+ :param dagbag: DagBag to retrieve from
+ :param dag_id: DAG ID to retrieve
+ :param task_id: Optional task ID to retrieve from the DAG
+ :return: tuple of (dag, task) where task is None if not requested
+ :raises ValueError: If DAG or task is not found
+ """
+ if dag_id not in dagbag.dags:
+ raise ValueError(
+ f"DAG '{dag_id}' not found in DagBag. "
+ f"This typically indicates a race condition where the DAG was removed or failed to parse."
+ )
+
+ dag = dagbag.dags[dag_id]
+
+ if task_id is not None:
+ try:
+ task = dag.get_task(task_id)
+ return dag, task
+ except TaskNotFound:
+ raise ValueError(
+ f"Task '{task_id}' not found in DAG '{dag_id}'. "
+ f"This typically indicates a race condition where the task was removed or the DAG structure changed."
Review Comment:
Ideally we shouldn't need those special exceptions for things like this:
https://lists.apache.org/thread/t8bnhyqy77kq4fk7fj3fmjd5wo9kv6w0
--
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] Improve DAG and task missing error handling in callbacks [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil merged PR #56725:
URL: https://github.com/apache/airflow/pull/56725
--
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] Improve DAG and task missing error handling in callbacks [airflow]
Posted by "rawwar (via GitHub)" <gi...@apache.org>.
rawwar commented on code in PR #56725:
URL: https://github.com/apache/airflow/pull/56725#discussion_r2438820355
##########
airflow-core/src/airflow/dag_processing/processor.py:
##########
@@ -234,6 +238,39 @@ def _serialize_dags(
return serialized_dags, serialization_import_errors
+def _get_dag_with_task(
+ dagbag: DagBag, dag_id: str, task_id: str | None = None
+) -> tuple[DAG, BaseOperator | MappedOperator | None]:
+ """
+ Retrieve a DAG and optionally a task from the DagBag.
+
+ :param dagbag: DagBag to retrieve from
+ :param dag_id: DAG ID to retrieve
+ :param task_id: Optional task ID to retrieve from the DAG
+ :return: tuple of (dag, task) where task is None if not requested
+ :raises ValueError: If DAG or task is not found
+ """
+ if dag_id not in dagbag.dags:
+ raise ValueError(
+ f"DAG '{dag_id}' not found in DagBag. "
+ f"This typically indicates a race condition where the DAG was removed or failed to parse."
+ )
+
+ dag = dagbag.dags[dag_id]
+
+ if task_id is not None:
+ try:
+ task = dag.get_task(task_id)
+ return dag, task
+ except TaskNotFound:
+ raise ValueError(
+ f"Task '{task_id}' not found in DAG '{dag_id}'. "
+ f"This typically indicates a race condition where the task was removed or the DAG structure changed."
Review Comment:
I agree about not using `AirflowException`. But, user defined exceptions can be helpful to maintain consistent error messages. And, I think, we should still have `DagNotFoundException`. But, this should just extend `Exception` instead of `AirflowException`. Wdyt?
--
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] Improve DAG and task missing error handling in callbacks [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #56725:
URL: https://github.com/apache/airflow/pull/56725#discussion_r2438009737
##########
airflow-core/src/airflow/dag_processing/processor.py:
##########
@@ -234,6 +238,39 @@ def _serialize_dags(
return serialized_dags, serialization_import_errors
+def _get_dag_with_task(
+ dagbag: DagBag, dag_id: str, task_id: str | None = None
+) -> tuple[DAG, BaseOperator | MappedOperator | None]:
+ """
+ Retrieve a DAG and optionally a task from the DagBag.
+
+ :param dagbag: DagBag to retrieve from
+ :param dag_id: DAG ID to retrieve
+ :param task_id: Optional task ID to retrieve from the DAG
+ :return: tuple of (dag, task) where task is None if not requested
+ :raises ValueError: If DAG or task is not found
+ """
+ if dag_id not in dagbag.dags:
+ raise ValueError(
+ f"DAG '{dag_id}' not found in DagBag. "
+ f"This typically indicates a race condition where the DAG was removed or failed to parse."
+ )
+
+ dag = dagbag.dags[dag_id]
+
+ if task_id is not None:
+ try:
+ task = dag.get_task(task_id)
+ return dag, task
+ except TaskNotFound:
+ raise ValueError(
+ f"Task '{task_id}' not found in DAG '{dag_id}'. "
+ f"This typically indicates a race condition where the task was removed or the DAG structure changed."
Review Comment:
https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#dont-raise-airflowexception-directly
--
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] Improve DAG and task missing error handling in callbacks [airflow]
Posted by "rawwar (via GitHub)" <gi...@apache.org>.
rawwar commented on code in PR #56725:
URL: https://github.com/apache/airflow/pull/56725#discussion_r2440838213
##########
airflow-core/src/airflow/dag_processing/processor.py:
##########
@@ -234,6 +238,39 @@ def _serialize_dags(
return serialized_dags, serialization_import_errors
+def _get_dag_with_task(
+ dagbag: DagBag, dag_id: str, task_id: str | None = None
+) -> tuple[DAG, BaseOperator | MappedOperator | None]:
+ """
+ Retrieve a DAG and optionally a task from the DagBag.
+
+ :param dagbag: DagBag to retrieve from
+ :param dag_id: DAG ID to retrieve
+ :param task_id: Optional task ID to retrieve from the DAG
+ :return: tuple of (dag, task) where task is None if not requested
+ :raises ValueError: If DAG or task is not found
+ """
+ if dag_id not in dagbag.dags:
+ raise ValueError(
+ f"DAG '{dag_id}' not found in DagBag. "
+ f"This typically indicates a race condition where the DAG was removed or failed to parse."
+ )
+
+ dag = dagbag.dags[dag_id]
+
+ if task_id is not None:
+ try:
+ task = dag.get_task(task_id)
+ return dag, task
+ except TaskNotFound:
+ raise ValueError(
+ f"Task '{task_id}' not found in DAG '{dag_id}'. "
+ f"This typically indicates a race condition where the task was removed or the DAG structure changed."
Review Comment:
I was just trying to defend the stance and by the time I wrote the last comment, I too felt it unnecessary
--
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] Improve DAG and task missing error handling in callbacks [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #56725:
URL: https://github.com/apache/airflow/pull/56725#discussion_r2435975604
##########
airflow-core/src/airflow/dag_processing/processor.py:
##########
@@ -234,6 +237,39 @@ def _serialize_dags(
return serialized_dags, serialization_import_errors
+def _get_dag_with_task(
+ dagbag: DagBag, dag_id: str, task_id: str | None = None
+) -> tuple[DAG, Operator | None]:
+ """
+ Retrieve a DAG and optionally a task from the DagBag.
+
+ :param dagbag: DagBag to retrieve from
+ :param dag_id: DAG ID to retrieve
+ :param task_id: Optional task ID to retrieve from the DAG
+ :return: tuple of (dag, task) where task is None if not requested
+ :raises ValueError: If DAG or task is not found
+ """
+ if dag_id not in dagbag.dags:
+ raise ValueError(
+ f"DAG '{dag_id}' not found in DagBag. "
+ f"This typically indicates a race condition where the DAG was removed or failed to parse."
+ )
+
+ dag = dagbag.dags[dag_id]
+
+ if task_id is not None:
+ try:
+ task = dag.get_task(task_id)
+ return dag, task
+ except TaskNotFound:
+ raise ValueError(
+ f"Task '{task_id}' not found in DAG '{dag_id}'. "
+ f"This typically indicates a race condition where the task was removed or the DAG structure changed."
Review Comment:
```suggestion
f"This typically indicates a race condition where the task was removed or the DAG structure changed mid execution."
```
--
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] Improve DAG and task missing error handling in callbacks [airflow]
Posted by "rawwar (via GitHub)" <gi...@apache.org>.
rawwar commented on code in PR #56725:
URL: https://github.com/apache/airflow/pull/56725#discussion_r2438005201
##########
airflow-core/src/airflow/dag_processing/processor.py:
##########
@@ -234,6 +238,39 @@ def _serialize_dags(
return serialized_dags, serialization_import_errors
+def _get_dag_with_task(
+ dagbag: DagBag, dag_id: str, task_id: str | None = None
+) -> tuple[DAG, BaseOperator | MappedOperator | None]:
+ """
+ Retrieve a DAG and optionally a task from the DagBag.
+
+ :param dagbag: DagBag to retrieve from
+ :param dag_id: DAG ID to retrieve
+ :param task_id: Optional task ID to retrieve from the DAG
+ :return: tuple of (dag, task) where task is None if not requested
+ :raises ValueError: If DAG or task is not found
+ """
+ if dag_id not in dagbag.dags:
+ raise ValueError(
+ f"DAG '{dag_id}' not found in DagBag. "
+ f"This typically indicates a race condition where the DAG was removed or failed to parse."
+ )
+
+ dag = dagbag.dags[dag_id]
+
+ if task_id is not None:
+ try:
+ task = dag.get_task(task_id)
+ return dag, task
+ except TaskNotFound:
+ raise ValueError(
+ f"Task '{task_id}' not found in DAG '{dag_id}'. "
+ f"This typically indicates a race condition where the task was removed or the DAG structure changed."
Review Comment:
@kaxil , just trying to understand why we are raising `ValueError` instead of `DagNotFound` or `TaskNotFound` 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] Improve DAG and task missing error handling in callbacks [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #56725:
URL: https://github.com/apache/airflow/pull/56725#discussion_r2436083780
##########
airflow-core/src/airflow/dag_processing/processor.py:
##########
@@ -234,6 +237,39 @@ def _serialize_dags(
return serialized_dags, serialization_import_errors
+def _get_dag_with_task(
+ dagbag: DagBag, dag_id: str, task_id: str | None = None
+) -> tuple[DAG, Operator | None]:
+ """
+ Retrieve a DAG and optionally a task from the DagBag.
+
+ :param dagbag: DagBag to retrieve from
+ :param dag_id: DAG ID to retrieve
+ :param task_id: Optional task ID to retrieve from the DAG
+ :return: tuple of (dag, task) where task is None if not requested
+ :raises ValueError: If DAG or task is not found
+ """
+ if dag_id not in dagbag.dags:
+ raise ValueError(
+ f"DAG '{dag_id}' not found in DagBag. "
+ f"This typically indicates a race condition where the DAG was removed or failed to parse."
+ )
+
+ dag = dagbag.dags[dag_id]
+
+ if task_id is not None:
+ try:
+ task = dag.get_task(task_id)
+ return dag, task
+ except TaskNotFound:
+ raise ValueError(
+ f"Task '{task_id}' not found in DAG '{dag_id}'. "
+ f"This typically indicates a race condition where the task was removed or the DAG structure changed."
Review Comment:
It doesn't have to be mid execution
--
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] Improve DAG and task missing error handling in callbacks [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #56725:
URL: https://github.com/apache/airflow/pull/56725#issuecomment-3411451342
### Backport failed to create: v3-1-test. View the failure log <a href='https://github.com/apache/airflow/actions/runs/18566394409'> Run details </a>
<table>
<tr>
<th>Status</th>
<th>Branch</th>
<th>Result</th>
</tr>
<tr>
<td>❌</td>
<td>v3-1-test</td>
<td><a href="https://github.com/apache/airflow/commit/c0f9d11ddb7ea11937ed2ca01f8e8bcfb467303b"><img src='https://img.shields.io/badge/Commit-c0f9d11-red' alt='Commit Link'></a></td>
</tr>
</table>
You can attempt to backport this manually by running:
```bash
cherry_picker c0f9d11 v3-1-test
```
This should apply the commit to the v3-1-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.
After you have resolved the conflicts, you can continue the backport process by running:
```bash
cherry_picker --continue
```
--
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