Restore pluggable email backend for task failure/retry alerts

airflow-core/newsfragments/69877.bugfix.rst

@@ -0,0 +1 @@
+Restore delivery of ``email_on_failure`` and ``email_on_retry`` task alerts through a custom ``[email] email_backend``. These alerts were routed unconditionally through ``SmtpNotifier``, so deployments using an Amazon SES, SendGrid or org-internal backend stopped receiving them. The default remains ``SmtpNotifier``. Note that a configured ``email_backend`` which cannot be imported now fails with a logged error instead of silently falling back to SMTP -- check that ``[email] email_backend`` still resolves before upgrading.

airflow-core/tests/unit/dag_processing/test_processor.py

@@ -24,9 +24,9 @@
 import textwrap
 import typing
 import uuid
-from collections.abc import Callable
+from collections.abc import Callable, Iterable
 from socket import socketpair
-from typing import TYPE_CHECKING, BinaryIO
+from typing import TYPE_CHECKING, Any, BinaryIO
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -1615,6 +1615,24 @@ def fake_collect_dags(self, *args, **kwargs):
             _execute_task_callbacks(dagbag, request, log)
 
 
+def _recording_email_backend(
+    to: list[str] | Iterable[str],
+    subject: str,
+    html_content: str,
+    files: list[str] | None = None,
+    dryrun: bool = False,
+    cc: str | Iterable[str] | None = None,
+    bcc: str | Iterable[str] | None = None,
+    mime_subtype: str = "mixed",
+    mime_charset: str = "utf-8",
+    conn_id: str | None = None,
+    custom_headers: dict[str, Any] | None = None,
+    **kwargs,
+) -> None:
+    """Legacy ``[email] email_backend`` stub, patched with a spec'd mock in tests."""
+    raise AssertionError("should be patched in the test")
+
+
 class TestExecuteEmailCallbacks:
     """Test the email callback execution functionality."""
 
@@ -1967,6 +1985,70 @@ def test_parse_file_passes_bundle_name_to_dagbag(self):
             call_kwargs = mock_dagbag_class.call_args.kwargs
             assert call_kwargs["bundle_name"] == "test_bundle"
 
+    def test_execute_email_callbacks_uses_custom_email_backend(self):
+        """The Dag-processor path honours a custom ``[email] email_backend``, like the worker path."""
+        backend = MagicMock(spec=_recording_email_backend)
+        dagbag = MagicMock(spec=DagBag)
+        with DAG(dag_id="test_dag") as dag:
+            BaseOperator(task_id="test_task", email=["test@example.com"])
+        dagbag.dags = {"test_dag": dag}
+
+        current_time = timezone.utcnow()
+        request = EmailRequest(
+            filepath="/path/to/dag.py",
+            bundle_name="test_bundle",
+            bundle_version="1.0.0",
+            ti=TIDataModel(
+                id=str(uuid.uuid4()),
+                task_id="test_task",
+                dag_id="test_dag",
+                run_id="test_run",
+                logical_date="2023-01-01T00:00:00Z",
+                try_number=1,
+                attempt_number=1,
+                state="failed",
+                dag_version_id=str(uuid.uuid4()),
+            ),
+            context_from_server=TIRunContext(
+                dag_run=DRDataModel(
+                    dag_id="test_dag",
+                    run_id="test_run",
+                    logical_date="2023-01-01T00:00:00Z",
+                    data_interval_start=current_time,
+                    data_interval_end=current_time,
+                    run_after=current_time,
+                    start_date=current_time,
+                    end_date=None,
+                    run_type="manual",
+                    state="running",
+                    consumed_asset_events=[],
+                    partition_key=None,
+                ),
+                max_tries=2,
+            ),
+            email_type="failure",
+            msg="Task failed",
+        )
+
+        conf_overrides = {
+            ("email", "email_backend"): f"{__name__}._recording_email_backend",
+            ("email", "email_conn_id"): "my_smtp",
+            ("email", "from_email"): "from@airflow",
+        }
+        with conf_vars(conf_overrides):
+            with patch(f"{__name__}._recording_email_backend", backend):
+                with patch(
+                    "airflow.providers.smtp.notifications.smtp.SmtpNotifier", autospec=True
+                ) as mock_smtp_notifier:
+                    _execute_email_callbacks(dagbag, request, MagicMock(spec=FilteringBoundLogger))
+
+        mock_smtp_notifier.assert_not_called()
+        backend.assert_called_once()
+        args, kwargs = backend.call_args
+        assert args[0] == ["test@example.com"]
+        assert kwargs["conn_id"] == "my_smtp"
+        assert kwargs["from_email"] == "from@airflow"
+
 
 class TestDagProcessingMessageTypes:
     def test_message_types_in_dag_processor(self):

task-sdk/src/airflow/sdk/execution_time/email_backend.py

@@ -0,0 +1,92 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Protocol
+
+from airflow.sdk.bases.notifier import BaseNotifier
+from airflow.sdk.configuration import conf
+from airflow.sdk.exceptions import AirflowConfigException
+
+if TYPE_CHECKING:
+    from collections.abc import Iterable
+
+    from airflow.sdk.definitions.context import Context
+
+_DEFAULT_EMAIL_BACKEND = "airflow.utils.email.send_email_smtp"
+
+
+class _ErrorEmailNotifier(Protocol):
+    """
+    Constructor contract shared by the notifiers used for failure and retry alerts.
+
+    ``BaseNotifier`` itself does not describe this -- its ``__init__`` takes only ``context`` --
+    so the shared shape is spelled out here to keep the call site type-checked.
+    """
+
+    def __call__(
+        self,
+        to: str | Iterable[str],
+        from_email: str | None = ...,
+        subject: str | None = ...,
+        html_content: str | None = ...,
+    ) -> BaseNotifier: ...
+
+
+class _LegacyEmailBackendNotifier(BaseNotifier):
+    """
+    Adapter that exposes a legacy ``[email] email_backend`` callable as a notifier.
+
+    Before failure and retry alerts were routed through ``BaseNotifier`` subclasses, deployments
+    configured them through ``[email] email_backend`` -- a callable with the
+    ``airflow.utils.email.send_email`` signature, such as the Amazon SES or SendGrid senders.
+    This adapter renders the standard email fields like any notifier, then loads and calls the
+    configured backend, so existing ``email_backend`` setups keep working unchanged.
+
+    The backend is resolved from config at notify time rather than imported statically, keeping
+    the Task SDK free of a hard dependency on ``airflow.utils.email`` (which lives in
+    ``airflow-core``).
+    """
+
+    template_fields = ("to", "from_email", "subject", "html_content")
+
+    def __init__(
+        self,
+        to: str | Iterable[str],
+        from_email: str | None = None,
+        subject: str | None = None,
+        html_content: str | None = None,
+        **kwargs: Any,
+    ) -> None:
+        super().__init__()
+        self.to = to
+        self.from_email = from_email
+        self.subject = subject
+        self.html_content = html_content
+
+    def notify(self, context: Context) -> None:
+        backend = conf.getimport("email", "email_backend", fallback=_DEFAULT_EMAIL_BACKEND)
+        if backend is None:
+            raise AirflowConfigException("`[email] email_backend` is not configured")
+        backend(
+            self.to,
+            self.subject,
+            self.html_content,
+            conn_id=conf.get("email", "email_conn_id", fallback=None),
+            from_email=self.from_email,
+        )

task-sdk/src/airflow/sdk/execution_time/task_runner.py

@@ -136,6 +136,11 @@
     get_previous_dagrun_success,
     set_current_context,
 )
+from airflow.sdk.execution_time.email_backend import (
+    _DEFAULT_EMAIL_BACKEND,
+    _ErrorEmailNotifier,
+    _LegacyEmailBackendNotifier,
+)
 from airflow.sdk.execution_time.sentry import Sentry
 from airflow.sdk.execution_time.xcom import XCom
 from airflow.sdk.listener import get_listener_manager
@@ -1998,20 +2003,39 @@ def _send_error_email_notification(
     error: BaseException | str | None,
     log: Logger,
 ) -> None:
-    """Send email notification for task errors using SmtpNotifier."""
-    try:
-        from airflow.providers.smtp.notifications.smtp import SmtpNotifier
-    except ImportError:
-        log.error(
-            "Failed to send task failure or retry email notification: "
-            "`apache-airflow-providers-smtp` is not installed. "
-            "Install this provider to enable email notifications."
-        )
-        return
+    """
+    Send email notification for task errors through the configured email backend.
+
+    A non-default ``[email] email_backend`` (an SES, SendGrid or org-internal callable with the
+    ``airflow.utils.email.send_email`` signature) is wrapped in
+    :class:`~airflow.sdk.execution_time.email_backend._LegacyEmailBackendNotifier`; otherwise the
+    default :class:`~airflow.providers.smtp.notifications.smtp.SmtpNotifier` is used.
 
+    Both the worker task-runner path (:func:`finalize`) and the DAG-processor callback path
+    (``_execute_email_callbacks``) funnel through this function, so the resolved backend is used
+    consistently regardless of how the task failed.
+    """
     if not task.email:
         return
 
+    email_backend = conf.get("email", "email_backend", fallback=_DEFAULT_EMAIL_BACKEND)
+    notifier_description = "SmtpNotifier"
+
+    if email_backend and email_backend != _DEFAULT_EMAIL_BACKEND:
+        notifier_class: _ErrorEmailNotifier = _LegacyEmailBackendNotifier
+        notifier_description = f"configured email_backend {email_backend!r}"
+    else:
+        try:
+            from airflow.providers.smtp.notifications.smtp import SmtpNotifier
+        except ImportError:
+            log.error(
+                "Failed to send task failure or retry email notification: "
+                "`apache-airflow-providers-smtp` is not installed. "
+                "Install this provider to enable email notifications."
+            )
+            return
+        notifier_class = SmtpNotifier
+
     subject_template_file = conf.get("email", "subject_template", fallback=None)
 
     # Read the template file if configured
@@ -2054,15 +2078,15 @@ def _send_error_email_notification(
         return
 
     try:
-        notifier = SmtpNotifier(
+        notifier = notifier_class(
             to=to_emails,
             subject=subject,
             html_content=html_content,
             from_email=conf.get("email", "from_email", fallback="airflow@airflow"),
         )
         notifier(email_context)
     except Exception:
-        log.exception("Failed to send email notification")
+        log.exception("Failed to send email notification via %s", notifier_description)
 
 
 @detail_span("task.execute")

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

@@ -0,0 +1,86 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from collections.abc import Iterable
+from typing import Any
+from unittest import mock
+
+import pytest
+
+from airflow.sdk.configuration import conf
+from airflow.sdk.exceptions import AirflowConfigException
+from airflow.sdk.execution_time.email_backend import _LegacyEmailBackendNotifier
+
+
+def _legacy_email_backend(
+    to: list[str] | Iterable[str],
+    subject: str,
+    html_content: str,
+    files: list[str] | None = None,
+    dryrun: bool = False,
+    cc: str | Iterable[str] | None = None,
+    bcc: str | Iterable[str] | None = None,
+    mime_subtype: str = "mixed",
+    mime_charset: str = "utf-8",
+    conn_id: str | None = None,
+    custom_headers: dict[str, Any] | None = None,
+    **kwargs,
+) -> None:
+    """
+    Spec for an ``[email] email_backend`` callable.
+
+    Mirrors the ``airflow.utils.email.send_email`` signature so the autospec enforces the
+    calling convention the notifier has to honour. Duplicated here rather than imported
+    because the Task SDK must not depend on ``airflow-core``.
+    """
+    raise AssertionError("spec only; never called")
+
+
+class TestLegacyEmailBackendNotifier:
+    def test_notify_calls_configured_backend(self):
+        """notify() loads the legacy backend from config and calls it with the standard fields."""
+        backend = mock.create_autospec(_legacy_email_backend)
+        notifier = _LegacyEmailBackendNotifier(
+            to=["a@b.com"],
+            from_email="from@x.com",
+            subject="Subject",
+            html_content="<p>body</p>",
+        )
+        with (
+            mock.patch.object(conf, "getimport", autospec=True, return_value=backend) as getimport,
+            mock.patch.object(conf, "get", autospec=True, return_value="my_conn"),
+        ):
+            notifier.notify(context={})
+
+        getimport.assert_called_once_with(
+            "email", "email_backend", fallback="airflow.utils.email.send_email_smtp"
+        )
+        backend.assert_called_once_with(
+            ["a@b.com"],
+            "Subject",
+            "<p>body</p>",
+            conn_id="my_conn",
+            from_email="from@x.com",
+        )
+
+    def test_notify_raises_when_backend_unresolvable(self):
+        """An empty/unloadable backend raises rather than silently doing nothing."""
+        notifier = _LegacyEmailBackendNotifier(to="a@b.com")
+        with mock.patch.object(conf, "getimport", autospec=True, return_value=None):
+            with pytest.raises(AirflowConfigException):
+                notifier.notify(context={})