jscheffl commented on code in PR #66400:
URL: https://github.com/apache/airflow/pull/66400#discussion_r3190340737
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py:
##########
@@ -681,22 +681,108 @@ def adopt_launched_task(
del tis_to_flush_by_key[ti_key]
self.running.add(ti_key)
+ def _alive_other_scheduler_job_ids(self) -> set[int]:
+ """
+ Return job IDs of every SchedulerJob that is currently alive —
excluding self.
+
+ "Alive" means ``Job.state == RUNNING`` AND its ``latest_heartbeat`` is
+ within ``[scheduler] scheduler_health_check_threshold``.
+
+ Used by ``_adopt_completed_pods`` to scope cross-scheduler pod
+ adoption to pods owned by no-longer-alive schedulers (#66396).
+ With a single scheduler the returned set is always empty — the
+ original "exclude self only" behavior is preserved. With multiple
+ schedulers each one only adopts pods whose owning scheduler is gone,
+ eliminating the relabel-thrash that PR #61839 introduced.
+
+ Returns an empty set on any DB error so the caller falls back to
+ the pre-#61839 "exclude self only" selector — a transient DB issue
+ must not break completed-pod cleanup.
+ """
+ if TYPE_CHECKING:
+ assert self.scheduler_job_id
+
+ try:
+ self_id = int(self.scheduler_job_id)
+ except (TypeError, ValueError):
+ # Tests sometimes set scheduler_job_id to a non-numeric string.
+ # In production it's always Job.id (int), but be defensive.
+ return set()
+
+ try:
+ from datetime import timedelta
+
+ from sqlalchemy import select
+
+ from airflow.jobs.job import Job
+ from airflow.utils import timezone
+ from airflow.utils.session import create_session
+ from airflow.utils.state import JobState
+
+ timeout = conf.getint("scheduler",
"scheduler_health_check_threshold")
+ cutoff = timezone.utcnow() - timedelta(seconds=timeout)
+ with create_session() as session:
+ rows = session.scalars(
+ select(Job.id).where(
+ Job.job_type == "SchedulerJob",
+ Job.state == JobState.RUNNING,
+ Job.latest_heartbeat >= cutoff,
+ Job.id != self_id,
+ )
+ ).all()
+ return set(rows)
Review Comment:
This call will retrieve all rows into a set - which might be very large
depending on paralellism.
Would it be potentially better to (1) not persist the full results into a
set but loop over the rows to keep memory foorprint lower and (2) also add a
`FOR UPDATE .. SKIP LOCKED` lock to prevent multiple schedulers in parallel
(assume there are three, one is unresponsive, two are running into this) are
not competing in parallel updating Pods? Else I fear this might generate new
conflicts
--
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]
