diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
index da4e1b54c94d3..ae4714d4b4e58 100644
--- a/airflow/models/taskinstance.py
+++ b/airflow/models/taskinstance.py
@@ -3087,6 +3087,10 @@ def handle_failure(
         if TYPE_CHECKING:
             assert self.task
             assert self.task.dag
+        try:
+            fail_stop = self.task.dag.fail_stop
+        except Exception:
+            fail_stop = False
         _handle_failure(
             task_instance=self,
             error=error,
@@ -3094,7 +3098,7 @@ def handle_failure(
             test_mode=test_mode,
             context=context,
             force_fail=force_fail,
-            fail_stop=self.task.dag.fail_stop,
+            fail_stop=fail_stop,
         )
 
     def is_eligible_to_retry(self):
diff --git a/airflow/serialization/pydantic/taskinstance.py b/airflow/serialization/pydantic/taskinstance.py
index 1d09b37a8dadc..2e01bf415a141 100644
--- a/airflow/serialization/pydantic/taskinstance.py
+++ b/airflow/serialization/pydantic/taskinstance.py
@@ -279,6 +279,10 @@ def handle_failure(
         if TYPE_CHECKING:
             assert self.task
             assert self.task.dag
+        try:
+            fail_stop = self.task.dag.fail_stop
+        except Exception:
+            fail_stop = False
         _handle_failure(
             task_instance=self,
             error=error,
@@ -286,7 +290,7 @@ def handle_failure(
             test_mode=test_mode,
             context=context,
             force_fail=force_fail,
-            fail_stop=self.task.dag.fail_stop,
+            fail_stop=fail_stop,
         )
 
     def refresh_from_task(self, task: Operator, pool_override: str | None = None) -> None: