dabla commented on code in PR #62922:
URL: https://github.com/apache/airflow/pull/62922#discussion_r3474124555
##########
task-sdk/src/airflow/sdk/execution_time/task_runner.py:
##########
@@ -2091,9 +2223,12 @@ def _execute_task(context: Context, ti:
RuntimeTaskInstance, log: Logger):
# Populate the context var so ExecutorSafeguard doesn't complain
ctx.run(ExecutorSafeguard.tracker.set, task)
- # Export context in os.environ to make it available for operators to use.
+ # Export context vars thread-safely to avoid race conditions in concurrent
execution
+ # (e.g., IterableOperator with AsyncAwareExecutor). Use
get_airflow_context_var() to read these.
+ # We intentionally do NOT update os.environ here as that would cause races
between tasks.
+ # We set these in the copied context so they're available inside
ctx.run(execute, ...).
airflow_context_vars = context_to_airflow_vars(context,
in_env_var_format=True)
- os.environ.update(airflow_context_vars)
+ ctx.run(_airflow_context_vars.set, airflow_context_vars)
Review Comment:
Resolve by doing a instanceof check on IndexedTaskInstance
--
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]
