hussein-awala commented on code in PR #32249:
URL: https://github.com/apache/airflow/pull/32249#discussion_r1265892884
##########
airflow/executors/kubernetes_executor_utils.py:
##########
@@ -129,7 +129,7 @@ def _run(
) -> str | None:
self.log.info("Event: and now my watch begins starting at
resource_version: %s", resource_version)
- kwargs = {"label_selector": f"airflow-worker={scheduler_job_id}"}
+ kwargs = {"label_selector": "airflow-worker"}
Review Comment:
By modifying the label selector to `airflow-worker`, the executor will watch
all pods created by Airflow. This change introduces a slight additional load,
but it is essential to address a critical bug in the executor that currently
disrupts its high-availability mode. IMO, this adjustment should not pose any
issues.
However, it is crucial to include a comment preceding this statement
clarifying the reason behind watching all pods. This comment will prevent
anyone from reverting the change under the false assumption that it improves
performance.
##########
airflow/executors/kubernetes_executor_utils.py:
##########
@@ -143,6 +143,10 @@ def _run(
self.log.debug("Event: %s had an event of type %s",
task.metadata.name, event["type"])
if event["type"] == "ERROR":
return self.process_error(event)
+ labels = task.metadata.labels
+ if labels.get("airflow-worker", None) != scheduler_job_id:
+ last_resource_version = task.metadata.resource_version
Review Comment:
Why do we need to update the `last_resource_version` when the pod is not
managed by this executor? To be safe, I think we should directly `continue`
when the label is not equals to the current scheduler job id. WDYT?
--
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]
