This is an automated email from the ASF dual-hosted git repository.
amoghrajesh 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 1e50b89cf86 Move retry policy decision outside post execute log group
block (#73023)
1e50b89cf86 is described below
commit 1e50b89cf8691a3acf0a60cf3a4f58d7fa6af2c8
Author: Amogh Desai <[email protected]>
AuthorDate: Tue Sep 15 09:05:24 2026 +0530
Move retry policy decision outside post execute log group block (#73023)
---
.../src/airflow/sdk/execution_time/task_runner.py | 2 ++
.../task_sdk/execution_time/test_task_runner.py | 36 ++++++++++++++++++++++
2 files changed, 38 insertions(+)
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 eeb27880622..e94958a8d3c 100644
--- a/task-sdk/src/airflow/sdk/execution_time/task_runner.py
+++ b/task-sdk/src/airflow/sdk/execution_time/task_runner.py
@@ -1815,6 +1815,8 @@ def _evaluate_retry_policy(
context=context,
)
if decision.reason:
+ # Close the group so the retry policy decision is not hidden
inside "Post Execute".
+ log.info("::endgroup::")
log.info("Retry policy decision", action=decision.action.value,
reason=decision.reason)
return decision
except Exception:
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 cc9fb77e089..17ba0709ca0 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
@@ -1271,6 +1271,42 @@ def
test_run_emits_post_execute_group_before_xcom_push(create_runtime_ti, mock_s
assert call_order.index("::group::Post Execute") <
call_order.index("Pushing xcom")
+def
test_retry_policy_decision_logged_outside_post_execute_group(create_runtime_ti,
mock_supervisor_comms):
+ """The retry policy decision log line closes the 'Post Execute' group
instead of nesting inside it."""
+ call_order: list[str] = []
+ tracked = {"::group::Post Execute", "::endgroup::", "Retry policy
decision"}
+
+ class _FailPolicy(RetryPolicy):
+ def evaluate(self, exception, try_number, max_tries, context=None):
+ return RetryDecision(action=RetryAction.FAIL, reason="auth error,
do not retry")
+
+ class _AlwaysFails(BaseOperator):
+ def execute(self, context):
+ raise RuntimeError("boom")
+
+ task = _AlwaysFails(task_id="fail_policy_task", retry_policy=_FailPolicy())
+ ti = create_runtime_ti(task=task, should_retry=True)
+ log = mock.MagicMock(spec=["info", "debug", "warning", "error",
"exception", "bind"])
+
+ def tracking_info(msg, *args, **kwargs):
+ if msg in tracked:
+ call_order.append(msg)
+
+ log.info.side_effect = tracking_info
+
+ state, msg, error = run(ti, context=ti.get_template_context(), log=log)
+ finalize(ti, state=state, context=ti.get_template_context(), log=log)
+
+ # No reopened "Post Execute", only finalize() ::endgroup:: follows the
retry policy decision.
+ assert call_order == [
+ "::endgroup::",
+ "::group::Post Execute",
+ "::endgroup::",
+ "Retry policy decision",
+ "::endgroup::",
+ ]
+
+
def test_finalize_emits_endgroup(create_runtime_ti, mock_supervisor_comms):
"""finalize() closes the post-execute log group but does not open it."""
task = BaseOperator(task_id="some_task")
