Check task attribute before use in add_tagging

airflow/sentry.py

@@ -124,7 +124,8 @@ def __init__(self):
         def add_tagging(self, task_instance):
             """Add tagging for a task_instance."""
             dag_run = task_instance.dag_run
-            task = task_instance.task
+            # See TaskInstance definition, the "task" attribute may not be set
+            task = getattr(task_instance, "task")
 
             with sentry_sdk.configure_scope() as scope:
                 for tag_name in self.SCOPE_TASK_INSTANCE_TAGS:
@@ -133,7 +134,8 @@ def add_tagging(self, task_instance):
                 for tag_name in self.SCOPE_DAG_RUN_TAGS:
                     attribute = getattr(dag_run, tag_name)
                     scope.set_tag(tag_name, attribute)
-                scope.set_tag("operator", task.__class__.__name__)
+                if task is not None:
+                    scope.set_tag("operator", task.__class__.__name__)
 
         @provide_session
         def add_breadcrumbs(