seanghaeli commented on code in PR #66608:
URL: https://github.com/apache/airflow/pull/66608#discussion_r3426203700
##########
task-sdk/src/airflow/sdk/execution_time/callback_supervisor.py:
##########
@@ -234,16 +257,39 @@ 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_MODULE_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,
)
+ # 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 to
surface the error rather than running degraded",
+ dag_id=dag_id,
+ run_id=run_id,
+ )
+ sys.exit(1)
Review Comment:
Agreed, closed the assymetry. I went with the "retry on the next loop"
approach on the executor side to match the triggerer side.
I think something more comprehensive might make sense in a future iteration,
like a proper retry policy, but for now I the gap is closed.
--
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]
