diff --git a/airflow-core/newsfragments/69877.bugfix.rst b/airflow-core/newsfragments/69877.bugfix.rst
new file mode 100644
index 0000000000000..7eef9afc99050
--- /dev/null
+++ b/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.
diff --git a/airflow-core/tests/unit/dag_processing/test_processor.py b/airflow-core/tests/unit/dag_processing/test_processor.py
index 3f219c1a083be..dd7f603b35bb0 100644
--- a/airflow-core/tests/unit/dag_processing/test_processor.py
+++ b/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):
diff --git a/task-sdk/src/airflow/sdk/execution_time/email_backend.py b/task-sdk/src/airflow/sdk/execution_time/email_backend.py
new file mode 100644
index 0000000000000..d307572e39d48
--- /dev/null
+++ b/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,
+ )
diff --git a/task-sdk/src/airflow/sdk/execution_time/task_runner.py b/task-sdk/src/airflow/sdk/execution_time/task_runner.py
index 94363b984b7f6..97d702e81253a 100644
--- a/task-sdk/src/airflow/sdk/execution_time/task_runner.py
+++ b/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,7 +2078,7 @@ def _send_error_email_notification(
return
try:
- notifier = SmtpNotifier(
+ notifier = notifier_class(
to=to_emails,
subject=subject,
html_content=html_content,
@@ -2062,7 +2086,7 @@ def _send_error_email_notification(
)
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")
diff --git a/task-sdk/tests/task_sdk/execution_time/test_email_backend.py b/task-sdk/tests/task_sdk/execution_time/test_email_backend.py
new file mode 100644
index 0000000000000..5940d0568c258
--- /dev/null
+++ b/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="body
",
+ )
+ 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",
+ "body
",
+ 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={})
diff --git a/task-sdk/tests/task_sdk/execution_time/test_task_runner.py b/task-sdk/tests/task_sdk/execution_time/test_task_runner.py
index 9d15ac40c6967..bee3f67b5a0ef 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_task_runner.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_task_runner.py
@@ -3842,6 +3842,30 @@ def mock_send_side_effect(*args, **kwargs):
)
+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 an autospecced mock in tests.
+
+ Mirrors the ``airflow.utils.email.send_email`` signature so the autospec enforces the
+ calling convention ``_LegacyEmailBackendNotifier`` has to honour. Duplicated here rather
+ than imported because the Task SDK must not depend on ``airflow-core``.
+ """
+ raise AssertionError("should be patched in the test")
+
+
class TestEmailNotifications:
FROM = "from@airflow"
@@ -4008,6 +4032,77 @@ def execute(self, context):
)
assert kwargs["from_email"] == self.FROM
+ def test_custom_email_backend_is_used(self, create_runtime_ti, mock_supervisor_comms):
+ """A custom ``[email] email_backend`` is wrapped and invoked with rendered fields."""
+ backend = mock.create_autospec(_recording_email_backend)
+
+ class FailingOperator(BaseOperator):
+ def execute(self, context):
+ raise AirflowFailException("Task failed on purpose")
+
+ task = FailingOperator(
+ task_id="legacy_backend_task",
+ email=["test@example.com"],
+ email_on_failure=True,
+ )
+
+ runtime_ti = create_runtime_ti(task=task)
+ context = runtime_ti.get_template_context()
+ log = mock.MagicMock()
+
+ with conf_vars(
+ {
+ ("email", "email_backend"): f"{__name__}._recording_email_backend",
+ ("email", "email_conn_id"): "my_smtp",
+ ("email", "from_email"): self.FROM,
+ }
+ ):
+ with mock.patch(f"{__name__}._recording_email_backend", backend):
+ with mock.patch(
+ "airflow.providers.smtp.notifications.smtp.SmtpNotifier", autospec=True
+ ) as mock_smtp_notifier:
+ state, _, error = run(runtime_ti, context, log)
+ finalize(runtime_ti, state, context, log, error)
+
+ # The default SMTP notifier must not be used when a custom backend is configured.
+ mock_smtp_notifier.assert_not_called()
+ backend.assert_called_once()
+ args, kwargs = backend.call_args
+ # send_email(to, subject, html_content, conn_id=..., from_email=...)
+ assert args[0] == ["test@example.com"]
+ assert kwargs["conn_id"] == "my_smtp"
+ assert kwargs["from_email"] == self.FROM
+
+ def test_unresolvable_email_backend_is_logged(self, create_runtime_ti, mock_supervisor_comms):
+ """An unimportable custom backend is logged, and does not silently fall back to SMTP."""
+
+ class FailingOperator(BaseOperator):
+ def execute(self, context):
+ raise AirflowFailException("Task failed on purpose")
+
+ task = FailingOperator(
+ task_id="bad_backend_task",
+ email=["test@example.com"],
+ email_on_failure=True,
+ )
+
+ runtime_ti = create_runtime_ti(task=task)
+ context = runtime_ti.get_template_context()
+ log = mock.MagicMock()
+
+ with conf_vars({("email", "email_backend"): "airflow.does.not.Exist"}):
+ # SmtpNotifier is patched to succeed, so a logged exception can only come from
+ # resolving the custom backend -- without that, this would pass pre-fix too.
+ with mock.patch(
+ "airflow.providers.smtp.notifications.smtp.SmtpNotifier", autospec=True
+ ) as mock_smtp_notifier:
+ state, _, error = run(runtime_ti, context, log)
+ # Must not raise even though the backend cannot be loaded.
+ finalize(runtime_ti, state, context, log, error)
+
+ mock_smtp_notifier.assert_not_called()
+ log.exception.assert_called()
+
@pytest.mark.enable_redact
def test_rendered_templates_mask_secrets(self, create_runtime_ti, mock_supervisor_comms):
"""Test that secrets registered with mask_secret() are redacted in rendered template fields."""