uranusjr commented on code in PR #68713:
URL: https://github.com/apache/airflow/pull/68713#discussion_r3489624512
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py:
##########
@@ -198,6 +198,64 @@ def start(self) -> None:
scheduler_job_id=self.scheduler_job_id,
)
+ def _coordinator_extra(self, queue: str | None) -> dict[str, Any] | None:
+ """
+ Return the ``extra`` mapping a coordinator declares for *queue*, if
any.
+
+ Read from the coordinator's declarative ``[sdk]`` config without
importing
+ or instantiating the coordinator. The coordinator manager only exists
on
+ Airflow 3.3+; on older Task SDKs the import fails and we fall back to
no
+ extra. A malformed ``[sdk] coordinators`` / ``queue_to_coordinator``
config
+ must not crash the scheduler on this first lookup either, so an invalid
+ config also falls back to no extra. The exception types are imported
from
+ ``airflow.sdk`` so they match whatever Task SDK actually raised them.
+ """
+ if not queue:
+ return None
+ try:
+ from airflow.sdk.exceptions import AirflowConfigException
+ from airflow.sdk.execution_time.coordinator import
get_coordinator_manager
+ except ImportError:
+ return None
+ try:
+ return get_coordinator_manager().extra_for_queue(queue)
+ except (AirflowConfigException, ValueError):
+ self.log.warning(
+ "Ignoring coordinator config for queue %s: invalid [sdk]
coordinator config",
+ queue,
+ exc_info=True,
+ )
+ return None
+
+ def _coordinator_pod_template_file(self, queue: str | None) -> str | None:
+ """
+ Return the pod template a coordinator declares for *queue*, if any.
+
+ Lets a queue routed to a non-Python coordinator (via ``[sdk]
+ queue_to_coordinator``) launch its worker pod from a
coordinator-specific
+ template — for example an image carrying the JVM for a Java
coordinator.
+ """
+ if (extra := self._coordinator_extra(queue)) is not None:
+ return extra.get("pod_template_file", None)
+ return None
+
+ def _coordinator_kube_image(self, queue: str | None) -> str | None:
+ """
+ Return the worker base image a coordinator declares for *queue*, if
any.
+
+ The base container image is never taken from a pod template; it comes
+ from ``kube_image``
(``worker_container_repository:worker_container_tag``)
+ or a per-task ``pod_override``. A coordinator may declare its own
+ ``worker_container_repository`` and ``worker_container_tag`` in
``extra``
+ (e.g. a JRE-bearing image for a Java coordinator); both are required to
+ compose an override, otherwise the executor default applies.
+ """
+ if (extra := self._coordinator_extra(queue)) is None:
Review Comment:
Both this and `_coordinator_pod_template_file` call `_coordinator_extra`.
Why not just call the function outside the two functions and pass the result
in? The `quque` value is only used to call the function anyway, not needed
otherwise.
--
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]
