You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "ephraimbuddy (via GitHub)" <gi...@apache.org> on 2026/04/20 08:06:24 UTC
Re: [PR] Guard 2 to 3+ migration against custom Dag bundle configs using startup-based initialization [airflow]
ephraimbuddy commented on code in PR #63185:
URL: https://github.com/apache/airflow/pull/63185#discussion_r3109123414
##########
airflow-core/src/airflow/dag_processing/bundles/manager.py:
##########
@@ -295,6 +296,76 @@ def _extract_and_sign_template(bundle_name: str) -> tuple[str | None, dict]:
session.execute(delete(ParseImportError).where(ParseImportError.bundle_name == name))
self.log.info("Deleted import errors for bundle %s which is no longer configured", name)
+ session.flush()
+
+ @provide_session
+ def reassign_dags_with_unconfigured_bundles(self, *, session: Session = NEW_SESSION) -> int:
+ """
+ Reassign Dags with unconfigured bundles to the first configured bundle.
+
+ This addresses Dags that reference bundles incorrectly (i.e. the Dag's
+ ``bundle_name`` matches a value in the database, but that database value
+ does not correspond to a user-configured bundle) as a side effect of the
+ ``0082_3_1_0_make_bundle_name_not_nullable.py`` migration
+ (see: https://github.com/apache/airflow/issues/63323).
+
+ Instead of attempting to infer the correct bundle for each Dag during the
+ migration, we reassign all Dags with unconfigured bundles to the first
+ configured bundle at DagFileProcessorManager startup. This relaxes the
+ "Requested bundle '{name}' is not configured." error that would otherwise
+ occur when triggering a DagRun immediately after the migration.
+
+ This fallback is not always semantically correct in environments using
+ multiple bundles, but it is a safe, temporary measure that allows users to
+ successfully trigger DagRuns right after the migration.
+
+ The correct Dag-to-bundle assignments will be restored by the Dag processor
+ on the next parsing cycle.
+
+ :param session: ORM Session
+ :return: Number of Dags reassigned.
+ """
+ # Import here to avoid circular import
+ # (manager -> dag -> dagrun -> taskinstance -> dag_version -> manager)
+ from airflow.models.dag import DagModel
+
+ configured_names = self.bundle_names
+ if not configured_names:
+ self.log.info(
+ "No Dag bundles are configured; skipping reassignment of Dags with unconfigured bundles."
+ )
+ return 0
+ default_bundle = configured_names[0]
+
+ count = cast(
+ "CursorResult",
+ session.execute(
+ update(DagModel)
+ .where(
+ or_(
+ DagModel.bundle_name.notin_(configured_names),
+ DagModel.bundle_name.is_(None),
+ )
+ )
+ .values(bundle_name=default_bundle)
+ .execution_options(synchronize_session="fetch")
Review Comment:
For 2.x upgraded DAGs, `relative_fileloc` can still be NULL because the migration only adds it as nullable. After this bulk reassignment, the first bundle refresh calls `deactivate_deleted_dags()`, where NULL is not in the observed file set, so these DAGs get marked stale before they are parsed. That defeats the “trigger immediately after migration” path and can leave DAGs hidden if the next parse fails. The fix needs to either populate/derive `relative_fileloc` for reassigned rows or keep legacy NULL rows out of stale deletion until a successful parse updates them.
##########
airflow-core/src/airflow/dag_processing/bundles/manager.py:
##########
@@ -295,6 +296,76 @@ def _extract_and_sign_template(bundle_name: str) -> tuple[str | None, dict]:
session.execute(delete(ParseImportError).where(ParseImportError.bundle_name == name))
self.log.info("Deleted import errors for bundle %s which is no longer configured", name)
+ session.flush()
+
+ @provide_session
+ def reassign_dags_with_unconfigured_bundles(self, *, session: Session = NEW_SESSION) -> int:
+ """
+ Reassign Dags with unconfigured bundles to the first configured bundle.
+
+ This addresses Dags that reference bundles incorrectly (i.e. the Dag's
+ ``bundle_name`` matches a value in the database, but that database value
+ does not correspond to a user-configured bundle) as a side effect of the
+ ``0082_3_1_0_make_bundle_name_not_nullable.py`` migration
+ (see: https://github.com/apache/airflow/issues/63323).
+
+ Instead of attempting to infer the correct bundle for each Dag during the
+ migration, we reassign all Dags with unconfigured bundles to the first
+ configured bundle at DagFileProcessorManager startup. This relaxes the
+ "Requested bundle '{name}' is not configured." error that would otherwise
+ occur when triggering a DagRun immediately after the migration.
+
+ This fallback is not always semantically correct in environments using
+ multiple bundles, but it is a safe, temporary measure that allows users to
+ successfully trigger DagRuns right after the migration.
+
+ The correct Dag-to-bundle assignments will be restored by the Dag processor
+ on the next parsing cycle.
+
+ :param session: ORM Session
+ :return: Number of Dags reassigned.
+ """
+ # Import here to avoid circular import
+ # (manager -> dag -> dagrun -> taskinstance -> dag_version -> manager)
+ from airflow.models.dag import DagModel
+
+ configured_names = self.bundle_names
Review Comment:
Reassign only to a synced bundle. `bundle_names` comes from `_bundle_config`, but `sync_bundles_to_db()` can skip a configured bundle when `get_bundle()` /bundle construction fails. If the first configured bundle fails, this update can write Dag rows to a bundle that was never inserted into `dag_bundle`, causing an FK failure and crashing Dag processor startup. The fallback should come from bundles that were actually persisted/active after sync.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org