vatsrahul1001 commented on code in PR #75010:
URL: https://github.com/apache/airflow/pull/75010#discussion_r4368032635
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -3994,6 +3994,11 @@ def _find_task_instances_without_heartbeats(self, *,
session: Session) -> list[T
session.execute(with_row_locks(run_query, of=DagRun,
session=session, skip_locked=True))
)
query = query.where(or_(~regional_expression(TI), tuple_(TI.dag_id,
TI.run_id).in_(locked_runs)))
+ # Bound the batch (oldest heartbeat first) so a mass heartbeat-loss
event does not purge every
+ # stale row, each with per-TI work under FOR UPDATE, in one
scheduler-stalling transaction.
+ query = query.order_by(TI.last_heartbeat_at)
+ if self.job.max_tis_per_query:
+ query = query.limit(self.job.max_tis_per_query)
Review Comment:
Good catch — fixed in 72d216c. The purge now returns `0` when it clears a
full batch, and `EventScheduler.call_regular_interval` reschedules the next
pass on the following loop instead of waiting the full detection interval. So a
mass event drains loop-paced, not interval-paced, while each pass stays bounded
— recovery is back to effectively one-pass-equivalent. And you're right that
`handle_failure()` commits per TI, so what the cap really bounds is
loop-occupancy per timer call; the comment now says that rather than implying
one giant transaction.
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -3994,6 +3994,11 @@ def _find_task_instances_without_heartbeats(self, *,
session: Session) -> list[T
session.execute(with_row_locks(run_query, of=DagRun,
session=session, skip_locked=True))
)
query = query.where(or_(~regional_expression(TI), tuple_(TI.dag_id,
TI.run_id).in_(locked_runs)))
+ # Bound the batch (oldest heartbeat first) so a mass heartbeat-loss
event does not purge every
+ # stale row, each with per-TI work under FOR UPDATE, in one
scheduler-stalling transaction.
+ query = query.order_by(TI.last_heartbeat_at)
Review Comment:
Agreed — fixed in 72d216c. That branch no longer `continue`s: a TI with no
resolvable `dag_version_id` is now failed via `handle_failure` (without the
callback), the same way a RESTARTING TI is released. This matches
`_ensure_ti_has_dag_version_id`'s own contract ('skip the callback', not 'skip
the TI') and removes the head-of-line starvation. Added
`test_purge_without_heartbeat_fails_ti_when_dag_version_unresolvable`.
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -3994,6 +3994,11 @@ def _find_task_instances_without_heartbeats(self, *,
session: Session) -> list[T
session.execute(with_row_locks(run_query, of=DagRun,
session=session, skip_locked=True))
)
query = query.where(or_(~regional_expression(TI), tuple_(TI.dag_id,
TI.run_id).in_(locked_runs)))
+ # Bound the batch (oldest heartbeat first) so a mass heartbeat-loss
event does not purge every
+ # stale row, each with per-TI work under FOR UPDATE, in one
scheduler-stalling transaction.
+ query = query.order_by(TI.last_heartbeat_at)
+ if self.job.max_tis_per_query:
Review Comment:
Fixed in 72d216c. The cap is now `self.job.max_tis_per_query or
self._parallelism`, so `0` maps to `[core] parallelism` (consistent with the
scheduling path) and the purge is never unbounded. Updated the `config.yml`
description to document the purge use and what `0` means for it. Covered by
`test_heartbeat_purge_batch_limit_falls_back_to_parallelism`.
##########
airflow-core/tests/unit/jobs/test_scheduler_job.py:
##########
@@ -918,6 +918,30 @@ def
test_heartbeat_scan_waits_for_a_locked_dagrun_only_for_regional_task_instanc
assert [row.id for row in
runner._find_task_instances_without_heartbeats(session=session)] == [ti.id]
+ def
test_find_task_instances_without_heartbeats_is_bounded_by_max_tis_per_query(self,
dag_maker, session):
+ """A mass heartbeat-loss event must be purged in bounded batches,
oldest heartbeat first."""
+ with dag_maker(serialized=True):
+ EmptyOperator(task_id="task")
+ runner = SchedulerJobRunner(Job(),
executors=[MockExecutor(do_update=False)])
+ session.add(runner.job)
+ session.flush()
+ runner.job.max_tis_per_query = 2
+
+ ids_oldest_first = []
+ for i in range(3):
+ dr = dag_maker.create_dagrun(run_id=f"run_{i}",
logical_date=DEFAULT_DATE + timedelta(days=i))
+ ti = dr.get_task_instance("task", session=session)
+ ti.state = State.RUNNING
+ ti.region_index = -1
+ ti.last_heartbeat_at = DEFAULT_DATE + timedelta(minutes=i)
Review Comment:
Fixed in bd1eca3b — the test now inserts heartbeats in reverse of insertion
order, so insertion/uuid7-id order differs from heartbeat order and the
assertion fails if the `order_by` is dropped. On the `max_tis_per_query = 0`
case: since `0` now maps to parallelism (there is no unbounded branch anymore),
that path is covered by
`test_heartbeat_purge_batch_limit_falls_back_to_parallelism`.
--
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]
