This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new cb9765fa4ee Don't retry task heartbeats according to ``[workers]
execution_api_retries` (#73244)
cb9765fa4ee is described below
commit cb9765fa4eed0ac4a19712d9c9c9d53d16292436
Author: Nick Stenning <[email protected]>
AuthorDate: Thu Sep 24 13:19:34 2026 +0200
Don't retry task heartbeats according to ``[workers] execution_api_retries`
(#73244)
Heartbeats had two stacked retry mechanisms: the Task SDK API client
retried every request, including heartbeats, and the supervisor
separately counted consecutive heartbeat failures against `[workers]
max_failed_heartbeats`. The client retries also had the unwanted
side-effect of blocking the supervisor's monitor loop while it backed
off, so during an API server outage the supervisor would be delayed
forwarding task logs and/or noticing that the task subprocess had
exited.
Heartbeat retries are now governed solely by `max_failed_heartbeats`.
I've adjusted the default value of that tunable from 3 to 30. This gives
a failure tolerance that (depending on the failure mode) is roughly
equivalent to the default value of `[scheduler]
task_instance_heartbeat_timeout`, and in any event slightly *more*
tolerant of an unreachable API server than the previous configuration.
Specifically, if every heartbeat failure is instant (e.g. a DNS lookup
failure or an explicitly rejected connection), then the new defaults
will result in task termination after 150s. If every heartbeat failure
takes a full 5s timeout, this extends to 300s, matching the scheduler's
tolerance for a missing task.
---
airflow-core/newsfragments/73244.significant.rst | 22 +++++++++++++
.../src/airflow/config_templates/config.yml | 8 ++++-
task-sdk/src/airflow/sdk/api/client.py | 37 ++++++++++++++++------
.../src/airflow/sdk/execution_time/supervisor.py | 3 +-
task-sdk/tests/task_sdk/api/test_client.py | 14 ++++++++
5 files changed, 72 insertions(+), 12 deletions(-)
diff --git a/airflow-core/newsfragments/73244.significant.rst
b/airflow-core/newsfragments/73244.significant.rst
new file mode 100644
index 00000000000..0c3c6e375c4
--- /dev/null
+++ b/airflow-core/newsfragments/73244.significant.rst
@@ -0,0 +1,22 @@
+Task heartbeats are no longer retried by ``[workers] execution_api_retries``
+
+Heartbeats had two overlapping retry mechanisms: the Task SDK API client
retried
+every request, including heartbeats, and the supervisor separately counted
+consecutive heartbeat failures against ``[workers] max_failed_heartbeats``. The
+client retries also had the unwanted side-effect of blocking the supervisor's
+monitor loop while it backed off, so during an API server outage the supervisor
+would be delayed forwarding task logs and/or noticing that the task subprocess
+had exited.
+
+Heartbeat retries are now governed solely by ``max_failed_heartbeats``, whose
+default has been adjusted from ``3`` to ``30`` to keep roughly the same
+tolerance for an unreachable API server as the two mechanisms combined gave
+before.
+
+If you have set ``[workers] max_failed_heartbeats`` explicitly, you may wish
+reconsider value you have chosen. It is now the whole retry budget: with the
+default ``[workers] min_heartbeat_interval`` of 5 seconds, the worker tolerates
+roughly ``5 * max_failed_heartbeats`` seconds of API unavailability before
+terminating a running task. You should likely keep the resulting value below
any
+configured ``[scheduler] task_instance_heartbeat_timeout`` (default 300
+seconds).
diff --git a/airflow-core/src/airflow/config_templates/config.yml
b/airflow-core/src/airflow/config_templates/config.yml
index 3cf451d55fc..4d3677a1d4a 100644
--- a/airflow-core/src/airflow/config_templates/config.yml
+++ b/airflow-core/src/airflow/config_templates/config.yml
@@ -2100,13 +2100,19 @@ workers:
max_failed_heartbeats:
description: |
The maximum number of consecutive failed heartbeats before terminating
the task instance process.
+
+ Unlike other calls to the execution API, heartbeats are not retried.
This setting and
+ ``min_heartbeat_interval`` together determine how long a worker
tolerates an unreachable API server
+ before killing running tasks.
version_added: 3.0.0
type: integer
example: ~
- default: "3"
+ default: "30"
execution_api_retries:
description: |
The maximum number of retry attempts to the execution API server.
+
+ Note that task instance heartbeats are not retried.
version_added: 3.0.0
type: integer
example: ~
diff --git a/task-sdk/src/airflow/sdk/api/client.py
b/task-sdk/src/airflow/sdk/api/client.py
index 11c170e5502..508807dface 100644
--- a/task-sdk/src/airflow/sdk/api/client.py
+++ b/task-sdk/src/airflow/sdk/api/client.py
@@ -339,7 +339,12 @@ class TaskInstanceOperations:
def heartbeat(self, id: uuid.UUID, pid: int):
body = TIHeartbeatInfo(pid=pid, hostname=get_hostname())
- self.client.put(f"task-instances/{id}/heartbeat",
content=body.model_dump_json())
+ self.client.request(
+ "PUT",
+ f"task-instances/{id}/heartbeat",
+ content=body.model_dump_json(),
+ retry=False, # heartbeats do not retry: they are implicitly
retried by the supervisor loop.
+ )
def skip_downstream_tasks(self, id: uuid.UUID, msg: SkipDownstreamTasks):
"""Tell the API server to skip the downstream tasks of this TI."""
@@ -1238,15 +1243,17 @@ class Client(httpx.Client):
log.debug("Execution API issued us a refreshed Task token")
self.auth = BearerAuth(new_token)
- @retry(
- retry=retry_if_exception(_should_retry_api_request),
- stop=stop_after_attempt(API_RETRIES),
- wait=wait_random_exponential(min=API_RETRY_WAIT_MIN,
max=API_RETRY_WAIT_MAX),
- before_sleep=_log_and_trace_retry,
- reraise=True,
- )
- def request(self, *args, **kwargs):
- """Implement a convenience for httpx.Client.request with a retry
layer."""
+ def request(self, *args, retry: bool = True, **kwargs):
+ """
+ Make a request using our httpx.Client.request with a default retry
policy.
+
+ Pass ``retry=False`` to bypass the default retry policy.
+ """
+ if not retry:
+ return self._request_without_retry(*args, **kwargs)
+ return self._request_with_retry(*args, **kwargs)
+
+ def _request_without_retry(self, *args, **kwargs):
# Set content type as convenience if not already set
if kwargs.get("content", None) is not None and "content-type" not in (
kwargs.get("headers", {}) or {}
@@ -1255,6 +1262,16 @@ class Client(httpx.Client):
return super().request(*args, **kwargs)
+ @retry(
+ retry=retry_if_exception(_should_retry_api_request),
+ stop=stop_after_attempt(API_RETRIES),
+ wait=wait_random_exponential(min=API_RETRY_WAIT_MIN,
max=API_RETRY_WAIT_MAX),
+ before_sleep=_log_and_trace_retry,
+ reraise=True,
+ )
+ def _request_with_retry(self, *args, **kwargs):
+ return self._request_without_retry(*args, **kwargs)
+
# We "group" or "namespace" operations by what they operate on, rather
than a flat namespace with all
# methods on one object prefixed with the object type
(`.task_instances.update` rather than
# `task_instance_update` etc.)
diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
index 42d1ddf8fe5..34863abb957 100644
--- a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
+++ b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
@@ -2387,7 +2387,8 @@ class InProcessTestSupervisor(ActivitySubprocess):
class _Client(Client):
def request(self, *args, **kwargs):
# Bypass the tenacity retries!
- return super().request.__wrapped__(self, *args, **kwargs) # type:
ignore[attr-defined]
+ kwargs["retry"] = False
+ return super().request(*args, **kwargs)
def _check_subprocess_exit(
self, raise_on_timeout: bool = False, expect_signal: None | int = None
diff --git a/task-sdk/tests/task_sdk/api/test_client.py
b/task-sdk/tests/task_sdk/api/test_client.py
index d3c47cebbf6..49690a2e07a 100644
--- a/task-sdk/tests/task_sdk/api/test_client.py
+++ b/task-sdk/tests/task_sdk/api/test_client.py
@@ -464,6 +464,20 @@ class TestTaskInstanceOperations:
client = make_client(transport=httpx.MockTransport(handle_request))
client.task_instances.heartbeat(ti_id, 100)
+ def test_task_instance_heartbeat_is_not_retried(self):
+ """A failing heartbeat surfaces at once so the supervisor's own retry
budget governs it."""
+ ti_id = uuid6.uuid7()
+ responses: list[httpx.Response] = [
+ httpx.Response(500, text="Internal Server Error"),
+ httpx.Response(204),
+ ]
+ client = make_client_w_responses(responses)
+
+ with pytest.raises(httpx.HTTPStatusError):
+ client.task_instances.heartbeat(ti_id, 100)
+
+ assert len(responses) == 1
+
@pytest.mark.parametrize("queues_enabled", [False, True])
def test_task_instance_defer(self, queues_enabled: bool):
# Simulate a successful response from the server that defers a task
