You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "AlexisBRENON (via GitHub)" <gi...@apache.org> on 2025/04/15 14:17:54 UTC
[I] DAG failure callback fails to execute when failing task has a mapped task downstream [airflow]
AlexisBRENON opened a new issue, #49297:
URL: https://github.com/apache/airflow/issues/49297
### Apache Airflow version
Other Airflow 2 version (please specify below)
### If "Other Airflow 2 version" selected, which one?
v2.9.3
### What happened?
I have a DAG with a first task generating a list of IDs to be processed by a downstream mapped task.
My whole DAG define a failure callback to notify me of errors.
When mapped tasks fails, I finally get the notification of failure, but if the first task fails, I don't get any notification. Actually, the callback method is never executed.
### What you think should happen instead?
Here is the scheduler logs when the first task succeed and the mapped tasks fail (expected behavior):
```
[2025-04-15T14:00:42.747+0000] {logging_mixin.py:188} INFO - [2025-04-15T14:00:42.747+0000] {dag.py:1541} INFO - Executing dag callback function: <function cb at 0x70b166ea22a0>
[2025-04-15T14:00:42.748+0000] {logging_mixin.py:188} INFO - [2025-04-15T14:00:42.748+0000] {test_task_map_cb.py:23} ERROR - Failure CB
```
Here is the same logs when the first task fails (downstream mapped tasks are marked as `upstream_failed`):
```txt
[2025-04-15T14:01:34.683+0000] {processor.py:715} ERROR - Error executing DagCallbackRequest callback for file: /opt/airflow/dags/greenbids/dags/test_task_map_cb.py
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/dag_processing/processor.py", line 713, in execute_callbacks
self._execute_dag_callbacks(dagbag, request, session)
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/utils/session.py", line 76, in wrapper
return func(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/dag_processing/processor.py", line 746, in _execute_dag_callbacks
callbacks, context = DAG.fetch_callback(
^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/api_internal/internal_api_call.py", line 115, in wrapper
return func(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/utils/session.py", line 76, in wrapper
return func(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/dag.py", line 1502, in fetch_callback
context = ti.get_template_context(session=session)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line 3027, in get_template_context
return _get_template_context(
^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line 752, in _get_template_context
expanded_ti_count: int | None = task.get_mapped_ti_count(task_instance.run_id, session=session)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/mappedoperator.py", line 797, in get_mapped_ti_count
current_count = self._get_specified_expand_input().get_total_map_length(run_id, session=session)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/expandinput.py", line 160, in get_total_map_length
lengths = self._get_map_lengths(run_id, session=session)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/expandinput.py", line 154, in _get_map_lengths
raise NotFullyPopulated(set(self.value).difference(map_lengths))
airflow.models.expandinput.NotFullyPopulated: Failed to populate all mapping metadata; missing: 'i_process'
```
### How to reproduce
You can use this DAG file to reproduce the issue:
```
import logging
from airflow.decorators import task, dag
from airflow.operators.python import get_current_context
from airflow.models.param import Param
_logger = logging.getLogger(__name__)
@task
def get_ids(total: int) -> list[int]:
if get_current_context().get("params", {}).get("fail_early", False):
raise RuntimeError("Test error")
return list(range(total))
@task
def process_id(i_process: int):
raise RuntimeError("Expected error reported")
def cb(*args, **kwargs):
_logger.error("Failure CB")
@dag(
dag_id="test.task_map",
schedule=None,
on_failure_callback=cb,
params={"fail_early": Param(False, type="boolean")},
)
def test_task_map():
ids = get_ids(5)
process_id.expand(i_process=ids)
test_task_map()
```
Trigger it a first time with it's parameter to its default value.
Then trigger it with the parameter set to True.
Compare the scheduler log in both cases.
### Operating System
Debian GNU/Linux 12 (bookworm) (custom Airflow image using Airflow 2.9.3-python3.12 as base)
### Versions of Apache Airflow Providers
_No response_
### Deployment
Official Apache Airflow Helm Chart
### Deployment details
Bug encountered in production instance deployed through official Helm chart on managed Google Kubernetes Engine.
Reproduced locally, while deploying with Docker compose.
### Anything else?
_No response_
### Are you willing to submit PR?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
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.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] DAG failure callback fails to execute when failing task has a mapped task downstream [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed issue #49297: DAG failure callback fails to execute when failing task has a mapped task downstream
URL: https://github.com/apache/airflow/issues/49297
--
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: [I] DAG failure callback fails to execute when failing task has a mapped task downstream [airflow]
Posted by "radumas (via GitHub)" <gi...@apache.org>.
radumas commented on issue #49297:
URL: https://github.com/apache/airflow/issues/49297#issuecomment-2937322188
I've been trying to debug a similar error in the logs so I thought this might be the issue in Airflow 3.0.1 but I can confirm the second run with `fail_early = True` produces
```
2025-06-03 17:49:36 [debug ] Sending [supervisor] msg=StartupDetails(ti=TaskInstance(id=UUID('019737c5-6d97-7939-8f17-db4113f8ffc8'), task_id='get_ids', dag_id='test.task_map', run_id='manual__2025-06-03T21:49:34.968030+00:00', try_number=1, map_index=-1, pool_slots=1, queue='default', priority_weight=2, executor_config={}, parent_context_carrier={}, context_carrier=None, queued_dttm=None), dag_rel_path='test_fail.py', bundle_info=BundleInfo(name='dags-folder', version=None), requests_fd=42, start_date=datetime.datetime(2025, 6, 3, 21, 49, 36, 325483, tzinfo=datetime.timezone.utc), ti_context=TIRunContext(dag_run=DagRun(dag_id='test.task_map', run_id='manual__2025-06-03T21:49:34.968030+00:00', logical_date=datetime.datetime(2025, 6, 3, 21, 49, 25, 939000, tzinfo=TzInfo(UTC)), data_interval_start=datetime.datetime(2025, 6, 3, 21, 49, 25, 939000, tzinfo=TzInfo(UTC)), data_interval_end=datetime.datetime(2025, 6, 3, 21, 49, 25, 939000, tzinfo=TzInfo(UTC)), ru
n_after=datetime.datetime(2025, 6, 3, 21, 49, 25, 939000, tzinfo=TzInfo(UTC)), start_date=datetime.datetime(2025, 6, 3, 21, 49, 36, 36358, tzinfo=TzInfo(UTC)), end_date=None, clear_number=0, run_type=<DagRunType.MANUAL: 'manual'>, conf={'fail_early': True}, consumed_asset_events=[]), task_reschedule_count=0, max_tries=0, variables=[], connections=[], upstream_map_indexes={}, next_method=None, next_kwargs=None, xcom_keys_to_clear=[], should_retry=False), type='StartupDetails')
2025-06-03 17:49:36 [debug ] Received message from task runner [supervisor] msg=SetRenderedFields(rendered_fields={'templates_dict': None, 'op_args': [5], 'op_kwargs': {}}, type='SetRenderedFields')
2025-06-03 17:49:36 [debug ] Received message from task runner [supervisor] msg=TaskState(state=<TaskInstanceState.FAILED: 'failed'>, end_date=datetime.datetime(2025, 6, 3, 21, 49, 36, 481092, tzinfo=TzInfo(UTC)), type='TaskState', rendered_map_index=None)
2025-06-03 17:49:36 [info ] Task finished [supervisor] duration=0.2048295510467142 exit_code=0 final_state=failed
--
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: [I] DAG failure callback fails to execute when failing task has a mapped task downstream [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on issue #49297:
URL: https://github.com/apache/airflow/issues/49297#issuecomment-4890684578
This issue has been closed because it has not received response from the issue author.
--
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: [I] DAG failure callback fails to execute when failing task has a mapped task downstream [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on issue #49297:
URL: https://github.com/apache/airflow/issues/49297#issuecomment-4637877999
This issue has been automatically marked as stale because it has been open for 365 days without any activity. There has been several Airflow releases since last activity on this issue. Kindly asking to recheck the report against latest Airflow version and let us know if the issue is reproducible. The issue will be closed in next 30 days if no further activity occurs from the issue author.
--
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