radumas opened a new issue, #51457: URL: https://github.com/apache/airflow/issues/51457
### Apache Airflow version 3.0.1 ### If "Other Airflow 2 version" selected, which one? _No response_ ### What happened? When dynamically generating taskgroups with a downstream task dependency. The downstream dependency will achieve a state mismatch `Executor LocalExecutor(parallelism=64) reported that the task instance <TaskInstance: cat_names_more_mapped_pipeline.final_task manual__2025-06-05T15:17:33.378964+00:00 [queued]> finished with state failed, but the task instance's state attribute is queued. Learn more: https://airflow.apache.org/docs/apache-airflow/stable/troubleshooting.html#task-state-changed-externally` And the `airflow-api-server` will generate a massive log that includes `airflow.sdk.definitions._internal.expandinput.NotFullyPopulated: Failed to populate all mapping metadata; missing: 'cat_name'` ### What you think should happen instead? The downstream task should run when upstream tasks are all done ### How to reproduce Ran the following DAG with "randomly" dynamically generated mapped tasks and a downstream task that waits for upstream tasks to succeed and comments on the tasks that were dynamically generated. Forgive the whimsy of the example, as well as some possible extraneous bits, I was trying to test various variables to reproduce the bug.  ```python from airflow import DAG import os import sys from random import randrange, choice from time import sleep from airflow.decorators import task, task_group from datetime import timedelta, datetime default_args = { "owner": "rdumas", "depends_on_past": False, "retry_delay": timedelta(minutes=3), } with DAG( dag_id="cat_names_more_mapped_pipeline", default_args=default_args, start_date=datetime(2025, 6, 1), schedule=None, catchup=False, max_active_runs=1, tags=["example", "cats"], ): @task def get_cat_names(): return ['LUNA','CHARLIE','BELLA','SHADOW','LUCY','COCO','LEO','MOLLY','SMOKEY','DAISY','LILY','OLIVER','TIGER','MAX','SIMBA','JACK','CLEO','BUDDY','MILO','OREO','LUCKY','PEPPER','GINGER','LOLA'] @task def filter_cats_by_greeting(cat_names): successful_greetings = [] for cat in cat_names: if choice([True, False]): # Randomized greeting success print(f"โ Greeted {cat}") successful_greetings.append(cat) else: print(f"โ {cat} ran away and wasn't greeted.") return successful_greetings @task_group(group_id="cat_actions") def cat_actions(cat_name): @task def feed_cat(cat_name): print(f"Feeding {cat_name.title()} some tuna...") return f"Fed {cat_name}" @task def play_with_cat(cat_name, **context): print(f"Playing with {cat_name.title()} using a laser pointer.") playtime=randrange(5,45) sleep(playtime) context["ti"].xcom_push(key="playtime", value=playtime) return f"Played with {cat_name}" play_with_cat(cat_name) >> feed_cat(cat_name) @task(trigger_rule="all_done" ) def final_task(greeted_cats, **context): ti = context["ti"] fed = ti.xcom_pull(task_ids="cat_actions.feed_cat", map_indexes=None) played = ti.xcom_pull(task_ids="cat_actions.play_with_cat", map_indexes=None) print("๐โโ๏ธ๐Greeted cats: \n", "\n".join(greeted_cats)) print("๐ฝ๏ธ Fed cats:", fed) print("๐พ Played with cats:", played) # Pipeline all_cats = get_cat_names() greeted_cats = filter_cats_by_greeting(all_cats) cat_actions.expand(cat_name=greeted_cats) >> final_task(greeted_cats) ``` log ``` INFO: 127.0.0.1:54190 - "PATCH /airflow/execution/task-instances/019740ab-3c6b-7c09-bfa0-956b7503ffc6/run HTTP/1.1" 500 Internal Server Error ERROR: Exception in ASGI application + Exception Group Traceback (most recent call last): | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/_utils.py", line 76, in collapse_excgroups | yield | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/base.py", line 181, in __call__ | recv_stream.close() | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/anyio/_backends/_asyncio.py", line 772, in __aexit__ | raise BaseExceptionGroup( | exceptiongroup.ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception) +-+---------------- 1 ---------------- | Traceback (most recent call last): | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/uvicorn/protocols/http/httptools_impl.py", line 409, in run_asgi | result = await app( # type: ignore[func-returns-value] | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/uvicorn/middleware/proxy_headers.py", line 60, in __call__ | return await self.app(scope, receive, send) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/fastapi/applications.py", line 1054, in __call__ | await super().__call__(scope, receive, send) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/applications.py", line 112, in __call__ | await self.middleware_stack(scope, receive, send) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/errors.py", line 187, in __call__ | raise exc | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/errors.py", line 165, in __call__ | await self.app(scope, receive, _send) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/gzip.py", line 29, in __call__ | await responder(scope, receive, send) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/gzip.py", line 126, in __call__ | await super().__call__(scope, receive, send) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/gzip.py", line 46, in __call__ | await self.app(scope, receive, self.send_with_compression) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/cors.py", line 85, in __call__ | await self.app(scope, receive, send) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/base.py", line 183, in __call__ | raise app_exc | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/base.py", line 141, in coro | await self.app(scope, receive_or_disconnect, send_no_error) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/exceptions.py", line 62, in __call__ | await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app | raise exc | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app | await app(scope, receive, sender) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/routing.py", line 714, in __call__ | await self.middleware_stack(scope, receive, send) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/routing.py", line 734, in app | await route.handle(scope, receive, send) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/routing.py", line 460, in handle | await self.app(scope, receive, send) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/cadwyn/applications.py", line 259, in __call__ | await self.__call__(scope, receive, send) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/fastapi/applications.py", line 1054, in __call__ | await super().__call__(scope, receive, send) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/applications.py", line 112, in __call__ | await self.middleware_stack(scope, receive, send) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/errors.py", line 187, in __call__ | raise exc | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/errors.py", line 165, in __call__ | await self.app(scope, receive, _send) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/base.py", line 181, in __call__ | recv_stream.close() | File "/usr/lib64/python3.9/contextlib.py", line 137, in __exit__ | self.gen.throw(typ, value, traceback) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/_utils.py", line 82, in collapse_excgroups | raise exc | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/base.py", line 178, in __call__ | response = await self.dispatch_func(request, call_next) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/cadwyn/middleware.py", line 124, in dispatch | response = await call_next(request) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/base.py", line 156, in call_next | raise app_exc | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/base.py", line 141, in coro | await self.app(scope, receive_or_disconnect, send_no_error) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/exceptions.py", line 62, in __call__ | await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app | raise exc | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app | await app(scope, receive, sender) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/cadwyn/routing.py", line 93, in __call__ | await self.process_request(scope=scope, receive=receive, send=send, routes=routes) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/cadwyn/routing.py", line 131, in process_request | await route.handle(scope, receive, send) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/routing.py", line 288, in handle | await self.app(scope, receive, send) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/routing.py", line 76, in app | await wrap_app_handling_exceptions(app, request)(scope, receive, send) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app | raise exc | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app | await app(scope, receive, sender) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/routing.py", line 73, in app | response = await f(request) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/fastapi/routing.py", line 301, in app | raw_response = await run_endpoint_function( | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/fastapi/routing.py", line 212, in run_endpoint_function | return await dependant.call(**values) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/cadwyn/structure/versions.py", line 474, in decorator | response = await self._convert_endpoint_response_to_version( | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/cadwyn/structure/versions.py", line 520, in _convert_endpoint_response_to_version | response_or_response_body: Union[FastapiResponse, object] = await run_in_threadpool( | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/concurrency.py", line 37, in run_in_threadpool | return await anyio.to_thread.run_sync(func) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/anyio/to_thread.py", line 56, in run_sync | return await get_async_backend().run_sync_in_worker_thread( | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/anyio/_backends/_asyncio.py", line 2470, in run_sync_in_worker_thread | return await future | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/anyio/_backends/_asyncio.py", line 967, in run | result = context.run(func, *args) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/cadwyn/schema_generation.py", line 511, in __call__ | return self._original_callable(*args, **kwargs) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/airflow/api_fastapi/execution_api/routes/task_instances.py", line 246, in ti_run | upstream_map_indexes = dict(_get_upstream_map_indexes(dag.get_task(ti.task_id), ti.map_index)) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/airflow/api_fastapi/execution_api/routes/task_instances.py", line 290, in _get_upstream_map_indexes | mapped_ti_count: int = upstream_task.task_group.get_parse_time_mapped_ti_count() | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/methodtools.py", line 72, in __call__ | return self.__call__(*args, **kwargs) | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/airflow/sdk/definitions/taskgroup.py", line 646, in get_parse_time_mapped_ti_count | return functools.reduce( | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/airflow/sdk/definitions/taskgroup.py", line 648, in <genexpr> | (g._expand_input.get_parse_time_mapped_ti_count() for g in self.iter_mapped_task_groups()), | File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/airflow/models/expandinput.py", line 75, in get_parse_time_mapped_ti_count | raise NotFullyPopulated(set(self.value).difference(literal_keys)) | airflow.sdk.definitions._internal.expandinput.NotFullyPopulated: Failed to populate all mapping metadata; missing: 'cat_name' +------------------------------------ During handling of the above exception, another exception occurred: Traceback (most recent call last): File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/uvicorn/protocols/http/httptools_impl.py", line 409, in run_asgi result = await app( # type: ignore[func-returns-value] File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/uvicorn/middleware/proxy_headers.py", line 60, in __call__ return await self.app(scope, receive, send) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/fastapi/applications.py", line 1054, in __call__ await super().__call__(scope, receive, send) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/applications.py", line 112, in __call__ await self.middleware_stack(scope, receive, send) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/errors.py", line 187, in __call__ raise exc File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/errors.py", line 165, in __call__ await self.app(scope, receive, _send) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/gzip.py", line 29, in __call__ await responder(scope, receive, send) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/gzip.py", line 126, in __call__ await super().__call__(scope, receive, send) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/gzip.py", line 46, in __call__ await self.app(scope, receive, self.send_with_compression) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/cors.py", line 85, in __call__ await self.app(scope, receive, send) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/base.py", line 183, in __call__ raise app_exc File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/base.py", line 141, in coro await self.app(scope, receive_or_disconnect, send_no_error) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/exceptions.py", line 62, in __call__ await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app raise exc File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app await app(scope, receive, sender) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/routing.py", line 714, in __call__ await self.middleware_stack(scope, receive, send) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/routing.py", line 734, in app await route.handle(scope, receive, send) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/routing.py", line 460, in handle await self.app(scope, receive, send) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/cadwyn/applications.py", line 259, in __call__ await self.__call__(scope, receive, send) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/fastapi/applications.py", line 1054, in __call__ await super().__call__(scope, receive, send) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/applications.py", line 112, in __call__ await self.middleware_stack(scope, receive, send) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/errors.py", line 187, in __call__ raise exc File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/errors.py", line 165, in __call__ await self.app(scope, receive, _send) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/base.py", line 181, in __call__ recv_stream.close() File "/usr/lib64/python3.9/contextlib.py", line 137, in __exit__ self.gen.throw(typ, value, traceback) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/_utils.py", line 82, in collapse_excgroups raise exc File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/base.py", line 178, in __call__ response = await self.dispatch_func(request, call_next) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/cadwyn/middleware.py", line 124, in dispatch response = await call_next(request) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/base.py", line 156, in call_next raise app_exc File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/base.py", line 141, in coro await self.app(scope, receive_or_disconnect, send_no_error) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/middleware/exceptions.py", line 62, in __call__ await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app raise exc File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app await app(scope, receive, sender) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/cadwyn/routing.py", line 93, in __call__ await self.process_request(scope=scope, receive=receive, send=send, routes=routes) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/cadwyn/routing.py", line 131, in process_request await route.handle(scope, receive, send) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/routing.py", line 288, in handle await self.app(scope, receive, send) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/routing.py", line 76, in app await wrap_app_handling_exceptions(app, request)(scope, receive, send) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app raise exc File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app await app(scope, receive, sender) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/routing.py", line 73, in app response = await f(request) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/fastapi/routing.py", line 301, in app raw_response = await run_endpoint_function( File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/fastapi/routing.py", line 212, in run_endpoint_function return await dependant.call(**values) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/cadwyn/structure/versions.py", line 474, in decorator response = await self._convert_endpoint_response_to_version( File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/cadwyn/structure/versions.py", line 520, in _convert_endpoint_response_to_version response_or_response_body: Union[FastapiResponse, object] = await run_in_threadpool( File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/starlette/concurrency.py", line 37, in run_in_threadpool return await anyio.to_thread.run_sync(func) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/anyio/to_thread.py", line 56, in run_sync return await get_async_backend().run_sync_in_worker_thread( File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/anyio/_backends/_asyncio.py", line 2470, in run_sync_in_worker_thread return await future File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/anyio/_backends/_asyncio.py", line 967, in run result = context.run(func, *args) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/cadwyn/schema_generation.py", line 511, in __call__ return self._original_callable(*args, **kwargs) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/airflow/api_fastapi/execution_api/routes/task_instances.py", line 246, in ti_run upstream_map_indexes = dict(_get_upstream_map_indexes(dag.get_task(ti.task_id), ti.map_index)) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/airflow/api_fastapi/execution_api/routes/task_instances.py", line 290, in _get_upstream_map_indexes mapped_ti_count: int = upstream_task.task_group.get_parse_time_mapped_ti_count() File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/methodtools.py", line 72, in __call__ return self.__call__(*args, **kwargs) File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/airflow/sdk/definitions/taskgroup.py", line 646, in get_parse_time_mapped_ti_count return functools.reduce( File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/airflow/sdk/definitions/taskgroup.py", line 648, in <genexpr> (g._expand_input.get_parse_time_mapped_ti_count() for g in self.iter_mapped_task_groups()), File "/home/airflow/airflow_venv/lib64/python3.9/site-packages/airflow/models/expandinput.py", line 75, in get_parse_time_mapped_ti_count raise NotFullyPopulated(set(self.value).difference(literal_keys)) airflow.sdk.definitions._internal.expandinput.NotFullyPopulated: Failed to populate all mapping metadata; missing: 'cat_name' ``` ### Operating System Red Hat Enterprise Linux 8.10 (Ootpa) ### Versions of Apache Airflow Providers _No response_ ### Deployment Virtualenv installation ### Deployment details _No response_ ### Anything else? This happens every time. I think it could be related to https://github.com/apache/airflow/issues/49377 but I didn't dig in to the testing DAG details. There could also be a relation to https://github.com/apache/airflow/issues/49297 but the example didn't reproduce in 3.0.1 ### Are you willing to submit PR? - [ ] 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
