This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new 639d37d5aa5 [v3-4-test] Let listeners see the end date of a skipped
task instance (#75685) (#75722)
639d37d5aa5 is described below
commit 639d37d5aa554743bf0871089a22b811b5dd8a93
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Oct 27 08:47:16 2026 +0000
[v3-4-test] Let listeners see the end date of a skipped task instance
(#75685) (#75722)
(cherry picked from commit bec92bcad025f0a424653d8ee4f47d39176cee70)
Co-authored-by: Kacper Muda <[email protected]>
---
task-sdk/src/airflow/sdk/execution_time/task_runner.py | 8 +++++---
task-sdk/tests/task_sdk/execution_time/test_task_runner.py | 9 ++++++++-
2 files changed, 13 insertions(+), 4 deletions(-)
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 ed1fa449e84..811ae4542d0 100644
--- a/task-sdk/src/airflow/sdk/execution_time/task_runner.py
+++ b/task-sdk/src/airflow/sdk/execution_time/task_runner.py
@@ -1712,9 +1712,10 @@ def _run_task_and_map_outcome(
log.info("::group::Post Execute")
if e.args:
log.info("Skipping task.", reason=e.args[0])
+ ti.end_date = datetime.now(tz=UTC)
msg = TaskState(
state=TaskInstanceState.SKIPPED,
- end_date=datetime.now(tz=UTC),
+ end_date=ti.end_date,
rendered_map_index=ti.rendered_map_index,
)
state = TaskInstanceState.SKIPPED
@@ -2108,6 +2109,7 @@ def _handle_trigger_dag_run(
)
if isinstance(comms_msg, ErrorResponse) and comms_msg.error ==
ErrorType.DAGRUN_ALREADY_EXISTS:
+ ti.end_date = datetime.now(tz=UTC)
if drte.skip_when_already_exists:
log.info(
"Dag Run already exists, skipping task as
skip_when_already_exists is set to True.",
@@ -2115,7 +2117,7 @@ def _handle_trigger_dag_run(
)
msg = TaskState(
state=TaskInstanceState.SKIPPED,
- end_date=datetime.now(tz=UTC),
+ end_date=ti.end_date,
rendered_map_index=ti.rendered_map_index,
)
state = TaskInstanceState.SKIPPED
@@ -2123,7 +2125,7 @@ def _handle_trigger_dag_run(
log.error("Dag Run already exists, marking task as failed.",
dag_id=drte.trigger_dag_id)
msg = TaskState(
state=TaskInstanceState.FAILED,
- end_date=datetime.now(tz=UTC),
+ end_date=ti.end_date,
rendered_map_index=ti.rendered_map_index,
)
state = TaskInstanceState.FAILED
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 c3f1b9e72e5..b8d9a555376 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
@@ -1997,6 +1997,7 @@ def test_run_basic_skipped(time_machine,
create_runtime_ti, mock_supervisor_comm
run(ti, context=ti.get_template_context(), log=mock.MagicMock())
assert ti.state == TaskInstanceState.SKIPPED
+ assert ti.end_date == instant
mock_supervisor_comms.send.assert_called_with(
TaskState(state=TaskInstanceState.SKIPPED, end_date=instant)
@@ -5419,6 +5420,7 @@ class TestTaskRunnerCallsListeners:
self.state = []
self.component = None
self.error = None
+ self.end_date = None
@hookimpl
def on_starting(self, component):
@@ -5440,6 +5442,7 @@ class TestTaskRunnerCallsListeners:
@hookimpl
def on_task_instance_skipped(self, previous_state, task_instance):
self.state.append(TaskInstanceState.SKIPPED)
+ self.end_date = task_instance.end_date
@hookimpl
def before_stopping(self, component):
@@ -5659,10 +5662,12 @@ class TestTaskRunnerCallsListeners:
)
log = mock.MagicMock()
context = runtime_ti.get_template_context()
- state, _, _ = run(runtime_ti, context, log)
+ state, msg, _ = run(runtime_ti, context, log)
finalize(runtime_ti, state, context, log)
assert listener.state == [TaskInstanceState.RUNNING,
TaskInstanceState.SKIPPED]
+ assert listener.end_date is not None
+ assert listener.end_date == msg.end_date
def test_listener_access_outlet_event_on_running_and_success(
self, mocked_parse, mock_supervisor_comms, listener_manager
@@ -6226,6 +6231,8 @@ class TestTriggerDagRunOperator:
assert state == expected_state
assert msg.state == expected_state
+ assert ti.end_date is not None
+ assert ti.end_date == msg.end_date
expected_calls = [
mock.call.send(
