[airflow] 27/38: Show serialization exceptions in DAG parsing log (#17277)

[jhtimmins]

This is an automated email from the ASF dual-hosted git repository.

jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git

commit db96ee5d6ac158fdb07f1d5a0cb66f0a15ff2aae
Author: Jed Cunningham <66968678+jedcunning...@users.noreply.github.com>
AuthorDate: Wed Jul 28 10:51:21 2021 -0600

    Show serialization exceptions in DAG parsing log (#17277)
    
    Make sure that any exceptions that happen when writing serialized DAGs
    to the db get written to the DAG parsing log, instead of only being added
    to `import_errors` for consumption via the UI.
    
    (cherry picked from commit 9cd5a97654fa82f1d4d8f599e8eb81957b3f7286)
---
 airflow/models/dagbag.py    | 1 +
 tests/models/test_dagbag.py | 4 +++-
 2 files changed, 4 insertions(+), 1 deletion(-)

diff --git a/airflow/models/dagbag.py b/airflow/models/dagbag.py
index 5d84ba2..b37482a 100644
--- a/airflow/models/dagbag.py
+++ b/airflow/models/dagbag.py
@@ -595,6 +595,7 @@ class DagBag(LoggingMixin):
             except OperationalError:
                 raise
             except Exception:
+                self.log.exception("Failed to write serialized DAG: %s", 
dag.full_filepath)
                 return [(dag.fileloc, 
traceback.format_exc(limit=-self.dagbag_import_error_traceback_depth))]
 
         # Retry 'DAG.bulk_write_to_db' & 'SerializedDagModel.bulk_sync_to_db' 
in case
diff --git a/tests/models/test_dagbag.py b/tests/models/test_dagbag.py
index b4edc0c..a065318 100644
--- a/tests/models/test_dagbag.py
+++ b/tests/models/test_dagbag.py
@@ -703,7 +703,9 @@ class TestDagBag(unittest.TestCase):
             )
             assert dagbag.import_errors == {}
 
-            dagbag.sync_to_db(session=session)
+            with self.assertLogs(level="ERROR") as cm:
+                dagbag.sync_to_db(session=session)
+            self.assertIn("SerializationError", "\n".join(cm.output))
 
             assert path in dagbag.import_errors
             err = dagbag.import_errors[path]