seanghaeli commented on code in PR #66608:
URL: https://github.com/apache/airflow/pull/66608#discussion_r3374918187
##########
task-sdk/src/airflow/sdk/execution_time/callback_supervisor.py:
##########
@@ -209,14 +218,36 @@ def _target():
_log.debug(
"Added bundle path to sys.path",
bundle_name=bundle_info.name, path=bundle_path
)
+ # DAG processor loads bundle files with a mangled module
name
+ # (unusual_prefix_{hash}_{stem}) to avoid collisions. The
callback path
+ # was serialized using that mangled name. Register the
module under that
+ # name so import_string can find it in the subprocess.
+ if callback_path and
callback_path.startswith("unusual_prefix_"):
+ _register_unusual_prefix_module(callback_path,
bundle.path, _log)
except Exception:
_log.warning(
"Failed to initialize DAG bundle for callback",
bundle_name=bundle_info.name,
exc_info=True,
)
- success, error_msg = execute_callback(callback_path,
callback_kwargs, _log)
+ # When DagRun identifiers are provided, fetch the DagRun via
SUPERVISOR_COMMS
+ # and build a context dict to pass to the callback function.
+ effective_kwargs = dict(callback_kwargs)
+ if dag_id and run_id:
+ context =
_fetch_and_build_context(task_runner.SUPERVISOR_COMMS, dag_id, run_id, _log)
+ if context is None:
+ _log.error(
+ "Cannot execute callback without context — failing so
it can be retried",
+ dag_id=dag_id,
+ run_id=run_id,
+ )
+ sys.exit(1)
Review Comment:
Updated the comment to be accurate — it now says "failing to surface the
error rather than running degraded."
On the retry question: path 1 (fetch before claiming) is what main does
today via `get_simple_context()`, but this PR moves context fetching to runtime
intentionally — fetching at queue time serializes the full DagRun context into
the callback's JSON blob in the DB, which doesn't scale (large contexts, stale
data if the run mutates between queuing and execution). Path 2 (retryable
result) has ambiguous scope — retry count, backoff, interaction with the
deadline's `missed` flag, and whether a permanently-unfetchable context should
retry forever all need design input. Terminal FAILED with an explicit error is
the right tradeoff for now: the user sees the failure in the UI and can
investigate vs running degraded with empty templates. Happy to follow up with
retry semantics as a separate discussion.
---
Drafted-by: Claude Code (Opus 4.6); reviewed by @seanghaeli before posting
--
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]
