This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new 1ebf5c9a424 [v3-3-test] Restore pluggable email backend for task
failure and retry alerts (#69877) (#70129)
1ebf5c9a424 is described below
commit 1ebf5c9a424b678d9b6dd273d3af7a999057274c
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Jul 21 16:50:44 2026 +0200
[v3-3-test] Restore pluggable email backend for task failure and retry
alerts (#69877) (#70129)
Since #57354, task email_on_failure / email_on_retry alerts were routed
unconditionally through SmtpNotifier, silently ignoring the
[email] email_backend configuration. Custom backends (SES, SendGrid,
org-internal) stopped delivering failure/retry emails even though the
deprecated email_on_* parameters still worked.
This restores the old behaviour using the existing [email] email_backend
option -- no new configuration is introduced:
- A non-default [email] email_backend is transparently wrapped in a new
LegacyEmailBackendNotifier, so existing SES / SendGrid / custom backends
keep delivering alerts unchanged. The backend is resolved from config at
notify time, so the Task SDK keeps no static dependency on
airflow.utils.email.
- Otherwise the default SmtpNotifier is used, exactly as before.
The notifier lives in the Task SDK
(airflow.sdk.execution_time.email_backend)
next to its only caller, so no extra provider needs to be installed for a
custom email backend to keep working.
Both failure-email entry points (the worker task-runner path and the
DAG-processor callback path) funnel through the same function, so the
selected backend is used consistently regardless of how the task failed.
The deprecated email_on_* parameters are not un-deprecated; this only keeps
their existing behaviour pluggable until removal in Airflow 4.
(cherry picked from commit f7dec02b2a9d88e886d0d3b2fd1919edbf2ff8af)
Co-authored-by: Jarek Potiuk <[email protected]>
---
airflow-core/newsfragments/69877.bugfix.rst | 1 +
.../tests/unit/dag_processing/test_processor.py | 86 +++++++++++++++++++-
.../airflow/sdk/execution_time/email_backend.py | 92 +++++++++++++++++++++
.../src/airflow/sdk/execution_time/task_runner.py | 48 ++++++++---
.../task_sdk/execution_time/test_email_backend.py | 86 ++++++++++++++++++++
.../task_sdk/execution_time/test_task_runner.py | 95 ++++++++++++++++++++++
6 files changed, 394 insertions(+), 14 deletions(-)
diff --git a/airflow-core/newsfragments/69877.bugfix.rst
b/airflow-core/newsfragments/69877.bugfix.rst
new file mode 100644
index 00000000000..7eef9afc990
--- /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 [...]
diff --git a/airflow-core/tests/unit/dag_processing/test_processor.py
b/airflow-core/tests/unit/dag_processing/test_processor.py
index 599c36d47f6..2da26aab124 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 sys
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
@@ -1571,6 +1571,24 @@ class TestExecuteTaskCallbacks:
_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."""
@@ -1923,6 +1941,70 @@ class TestExecuteEmailCallbacks:
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=["[email protected]"])
+ 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] == ["[email protected]"]
+ 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 00000000000..d307572e39d
--- /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 7a77ed9ad23..3200edf0db4 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 @@ from airflow.sdk.execution_time.context import (
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
@@ -1997,20 +2002,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
@@ -2053,7 +2077,7 @@ def _send_error_email_notification(
return
try:
- notifier = SmtpNotifier(
+ notifier = notifier_class(
to=to_emails,
subject=subject,
html_content=html_content,
@@ -2061,7 +2085,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 00000000000..5940d0568c2
--- /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=["[email protected]"],
+ from_email="[email protected]",
+ 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(
+ ["[email protected]"],
+ "Subject",
+ "<p>body</p>",
+ conn_id="my_conn",
+ from_email="[email protected]",
+ )
+
+ def test_notify_raises_when_backend_unresolvable(self):
+ """An empty/unloadable backend raises rather than silently doing
nothing."""
+ notifier = _LegacyEmailBackendNotifier(to="[email protected]")
+ 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 1aad1a02d5e..9714a49e872 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
@@ -3766,6 +3766,30 @@ class TestXComAfterTaskExecution:
)
+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"
@@ -3932,6 +3956,77 @@ class TestEmailNotifications:
)
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=["[email protected]"],
+ 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] == ["[email protected]"]
+ 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=["[email protected]"],
+ 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."""
