ephraimbuddy commented on code in PR #63692:
URL: https://github.com/apache/airflow/pull/63692#discussion_r3084782637
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -561,6 +561,15 @@ def _fetch_callbacks(
]
for callback in callbacks:
req = callback.get_callback_request()
+ if req.bundle_name not in bundle_names:
+ self.log.warning(
+ "Callback for dag_id=%s has bundle_name=%r which is
not served by "
+ "this DAG processor (serving bundles: %s). Skipping.",
+ getattr(req, "dag_id", "unknown"),
+ req.bundle_name,
+ bundle_names,
Review Comment:
Any callback fetched at this point is bound to have bundle_name in
bundle_names. Debugging here?
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -1331,16 +1331,26 @@ def produce_dag_callback(
) -> DagCallbackRequest | None:
"""Create a callback request for the DAG, or execute the callbacks
directly if instructed, and return None."""
if not execute:
+ try:
+ context_from_server = DagRunContext(
+ dag_run=self,
+ last_ti=relevant_ti,
+ )
+ except Exception:
+ self.log.exception(
+ "Failed to build DagRunContext for dag_id=%s run_id=%s; "
+ "sending callback with minimal context",
+ self.dag_id,
+ self.run_id,
+ )
+ context_from_server = None
Review Comment:
Even if it happens, this is bad: running callback without the full context
is worse than the callback failing. Also, the exception is broad here
--
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]
