hkc-8010 commented on code in PR #68008:
URL: https://github.com/apache/airflow/pull/68008#discussion_r3621371667
##########
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:
Done in b8bcab0c76c. This test now uses a real
`MockExecutor(do_update=False)` instead of the bare `MagicMock()`, matching the
other heartbeat-timeout tests, and asserts on `callback_sink.send`.
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -187,6 +187,26 @@ def _eager_load_dag_run_for_validation() ->
tuple[LoaderOption, LoaderOption]:
)
+def _resolve_ti_callback_bundle_info(ti: TaskInstance) -> tuple[str, str |
None, Any]:
+ """
+ Resolve the bundle name/version/version-data needed to build a
TaskCallbackRequest or EmailRequest.
+
+ Shared by the executor-reported "task finished but TI still looked queued"
path and the
+ heartbeat-timeout purge path, both of which need identical bundle-pinning
semantics: fall back
+ to ``dag_model`` for legacy tasks with no ``dag_version`` (pre-AIP-66
migrations), and leave the
+ bundle version unpinned when the dag run itself wasn't pinned
(``disable_bundle_versioning``),
+ so the callback runs against the same code as the task did.
+ """
Review Comment:
Good point, reworded in b8bcab0c76c. The docstring now says the helper is
used by the heartbeat-timeout purge path and that `process_executor_events`
inlines the same resolution rather than sharing it.
--
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]
