fix: task run exception never catch by Sentry

task-sdk/src/airflow/sdk/execution_time/sentry/configured.py

@@ -144,10 +144,14 @@ def wrapped_run(ti: RuntimeTaskInstance, context: Context, log: Logger) -> RunRe
                 try:
                     self.add_tagging(context["dag_run"], ti)
                     self.add_breadcrumbs(ti)
-                    return run(ti, context, log)
+                    run_return = run(ti, context, log)
                 except Exception as e:
                     sentry_sdk.capture_exception(e)
                     raise
+            _, _, run_error = run_return
+            if run_error:
+                sentry_sdk.capture_exception(run_error)
+            return run_return
 
         return wrapped_run
 

task-sdk/tests/task_sdk/execution_time/test_sentry.py

@@ -27,6 +27,7 @@
 import uuid6
 
 from airflow.providers.standard.operators.python import PythonOperator
+from airflow.sdk import Context, get_current_context
 from airflow.sdk._shared.module_loading import import_string
 from airflow.sdk._shared.timezones import timezone
 from airflow.sdk.api.datamodels._generated import DagRun, DagRunState, DagRunType, TaskInstanceState
@@ -35,6 +36,8 @@
 
 from tests_common.test_utils.config import conf_vars
 
+from structlog.typing import FilteringBoundLogger as Logger
+
 LOGICAL_DATE = timezone.utcnow()
 SCHEDULE_INTERVAL = datetime.timedelta(days=1)
 DATA_INTERVAL = (LOGICAL_DATE, LOGICAL_DATE + SCHEDULE_INTERVAL)
@@ -123,6 +126,7 @@ def mock_sentry_sdk(self):
         sentry_sdk.integrations = mock.Mock(logging=sentry_sdk_integrations_logging)
         sentry_sdk.configure_scope = mock.MagicMock()
         sentry_sdk.add_breadcrumb = mock.MagicMock()
+        sentry_sdk.capture_exception = mock.MagicMock()
 
         sys.modules["sentry_sdk"] = sentry_sdk
         sys.modules["sentry_sdk.integrations.logging"] = sentry_sdk_integrations_logging
@@ -270,3 +274,45 @@ def test_minimum_config(self, mock_sentry_sdk, sentry_minimum):
         sentry_minimum.prepare_to_enrich_errors(executor_integration="")
         assert mock_sentry_sdk.integrations.logging.ignore_logger.mock_calls == [mock.call("airflow.task")]
         assert mock_sentry_sdk.init.mock_calls == [mock.call(integrations=[])]
+
+    @pytest.mark.parametrize(
+        ('run_exception_return', 'run_raise'),
+        (
+                pytest.param(
+                    ValueError("This is Run Exception"),
+                    False,
+                    id="run_with_raise_exception"
+                ),
+                pytest.param(
+                    None,
+                    True,
+                    id="run_with_return_exception"
+                ),
+                pytest.param(
+                    ValueError("This is Run Exception"),
+                    True,
+                    id="run_with_return_and_raise_exception"
+                ),
+                pytest.param(
+                    None,
+                    False,
+                    id="run_without_exception"
+                )
+        )
+    )
+    def test_sentry_capture_exception(self, mock_sentry_sdk, task_instance, run_exception_return, run_raise):
+        """
+        Test that sentry_sdk.capture_exception is called on error
+        """
+        def run_side_effect(ti: RuntimeTaskInstance, context: Context, log: Logger):
+            if run_raise:
+                raise ValueError("This is Run Exception")
+            return STATE, None, run_exception_return
+
+        log = mock.Mock()
+
+        with mock.patch("airflow.sdk.execution_time.task_runner.run", side_effect=run_side_effect) as run:
+            run(task_instance, get_current_context(), log)
+
+        if run_exception_return is not None or run_raise:
+            mock_sentry_sdk.capture_exception.assert_called()