hkc-8010 commented on code in PR #68008:
URL: https://github.com/apache/airflow/pull/68008#discussion_r3510902710
##########
airflow-core/tests/unit/jobs/test_scheduler_job.py:
##########
@@ -8308,6 +8308,38 @@ def
test_scheduler_passes_context_from_server_on_heartbeat_timeout(self, dag_mak
assert callback_request.context_from_server.dag_run.logical_date ==
dag_run.logical_date
assert callback_request.context_from_server.max_tries == ti.max_tries
+ def test_heartbeat_timeout_converges_ti_state_before_next_scan(self,
dag_maker, session):
+ """A heartbeat-timed-out TI should not be found again on the next
scheduler scan."""
+ with dag_maker(dag_id="test_heartbeat_timeout_dedupe",
session=session):
+ EmptyOperator(task_id="test_task", on_failure_callback=lambda
context: None)
Review Comment:
Added a separate parametrized test,
`test_heartbeat_timeout_sets_callback_type_param`, that covers the
retry-eligible case (`max_tries=1`, `on_retry_callback`) alongside a
non-eligible case. Left this test's `max_tries=0` as is, since it is
specifically asserting the dedupe/converges-before-next-scan behavior, not
callback type.
##########
airflow-core/tests/unit/jobs/test_scheduler_job.py:
##########
@@ -8308,6 +8308,38 @@ def
test_scheduler_passes_context_from_server_on_heartbeat_timeout(self, dag_mak
assert callback_request.context_from_server.dag_run.logical_date ==
dag_run.logical_date
assert callback_request.context_from_server.max_tries == ti.max_tries
+ def test_heartbeat_timeout_converges_ti_state_before_next_scan(self,
dag_maker, session):
+ """A heartbeat-timed-out TI should not be found again on the next
scheduler scan."""
+ with dag_maker(dag_id="test_heartbeat_timeout_dedupe",
session=session):
+ EmptyOperator(task_id="test_task", on_failure_callback=lambda
context: None)
+
+ dag_run = dag_maker.create_dagrun(run_id="test_run",
state=DagRunState.RUNNING)
+
+ mock_executor = MagicMock()
Review Comment:
Used a real `MockExecutor(do_update=False)` in the new test instead of a
bare `MagicMock()`, matching the pattern already used in
`test_external_kill_sets_callback_type_param` for the equivalent scenario. Left
the existing test's `MagicMock()` alone since it only asserts
`send_callback.assert_called_once()`.
##########
airflow-core/tests/unit/jobs/test_scheduler_job.py:
##########
@@ -8308,6 +8308,38 @@ def
test_scheduler_passes_context_from_server_on_heartbeat_timeout(self, dag_mak
assert callback_request.context_from_server.dag_run.logical_date ==
dag_run.logical_date
assert callback_request.context_from_server.max_tries == ti.max_tries
+ def test_heartbeat_timeout_converges_ti_state_before_next_scan(self,
dag_maker, session):
Review Comment:
Added `test_heartbeat_timeout_sets_callback_type_param`, parametrized over a
retry-eligible case (`max_tries=1`) and a failure case (`max_tries=0`). It
verifies both `task_callback_type == UP_FOR_RETRY` on the callback request and
that the TI ends in `UP_FOR_RETRY` (not purged into `FAILED`) after the
heartbeat-timeout purge.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
