shahar1 commented on code in PR #49899:
URL: https://github.com/apache/airflow/pull/49899#discussion_r2192614821
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/job.py:
##########
@@ -168,34 +176,43 @@ def execute(self, context: Context):
ti.xcom_push(key="job_namespace", value=self.job.metadata.namespace)
self.pod: k8s.V1Pod | None
- if self.pod is None:
- self.pod = self.get_or_create_pod( # must set `self.pod` for
`on_kill`
+ self.pods: Sequence[k8s.V1Pod] | None = None
+ if self.parallelism is None and self.pod is None:
+ self.pod = self.get_or_create_pod(
pod_request_obj=self.pod_request_obj,
context=context,
)
+ self.pods = [
+ self.pod,
+ ]
Review Comment:
```suggestion
self.pods = [self.get_or_create_pod(
pod_request_obj=self.pod_request_obj,
context=context,
)
]
```
(please validate proper linting)
Also, please ensure that `self.pod` is being deprecated properly, i.e, if
being accessed - raise `AirflowProviderDeprecationWarning` + return the value
of `self.pods[0]`. Conceptual implementation (please modify as necessary and
add tests):
```python
@property
def pod(self):
warnings.warn("...", AirflowProviderDeprecationWarning, stacklevel=2)
return self.pods[0]
```
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/job.py:
##########
@@ -400,6 +419,29 @@ def reconcile_job_specs(
return None
+ def get_pods(
+ self, pod_request_obj: k8s.V1Pod, context: Context, *,
exclude_checked: bool = True
+ ) -> Sequence[k8s.V1Pod]:
+ """Return an already-running pods if exists."""
+ label_selector = self._build_find_pod_label_selector(context,
exclude_checked=exclude_checked)
+ pod_list: Sequence[k8s.V1Pod] = []
+ retry_number: int = 0
+
+ while len(pod_list) != self.parallelism or retry_number <= 3:
Review Comment:
Could you please make the max `retry_number` as a parameter?
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/triggers/job.py:
##########
@@ -69,7 +73,18 @@ def __init__(
super().__init__()
self.job_name = job_name
self.job_namespace = job_namespace
- self.pod_name = pod_name
+ if pod_name is not None:
+ warnings.warn(
+ "`pod_name` parameter is deprecated, please use `pod_names`",
+ AirflowProviderDeprecationWarning,
+ stacklevel=2,
+ )
+ self.pod_name = pod_name
Review Comment:
```suggestion
```
Instead of initializing it here, make `self.pod_name` a read-only property
(see previous suggestion for `self.pod` of the operator).
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/job.py:
##########
@@ -232,20 +249,23 @@ def execute_complete(self, context: Context, event: dict,
**kwargs):
raise AirflowException(event["message"])
if self.get_logs:
- pod_name = event["pod_name"]
- pod_namespace = event["pod_namespace"]
- self.pod = self.hook.get_pod(pod_name, pod_namespace)
- if not self.pod:
- raise PodNotFoundException("Could not find pod after resuming
from deferral")
- self._write_logs(self.pod)
+ for pod_name in event["pod_names"]:
+ pod_namespace = event["pod_namespace"]
+ pod = self.hook.get_pod(pod_name, pod_namespace)
+ if not pod:
+ raise PodNotFoundException("Could not find pod after
resuming from deferral")
+ self._write_logs(pod)
if self.do_xcom_push:
- xcom_result = event["xcom_result"]
- if isinstance(xcom_result, str) and xcom_result.rstrip() ==
EMPTY_XCOM_RESULT:
- self.log.info("xcom result file is empty.")
- return None
- self.log.info("xcom result: \n%s", xcom_result)
- return json.loads(xcom_result)
+ xcom_results: list[Any | None] = []
+ for xcom_result in event["xcom_result"]:
+ if isinstance(xcom_result, str) and xcom_result.rstrip() ==
EMPTY_XCOM_RESULT:
+ self.log.info("xcom result file is empty.")
+ xcom_results.append(None)
+ continue
+ self.log.info("xcom result: \n%s", xcom_result)
+ xcom_results.append(json.loads(xcom_result))
+ return xcom_results[0] if len(xcom_results) == 1 else xcom_results
Review Comment:
If returning a single entity is for retaining backward compatibilty and/or
nicer formatting - I'd maybe make it boolean flag as default, e.g.,. :
```python
return xcom_results[0] if self.unwrap_single and
len(xcom_results) == 1 else xcom_results # self.unwrap_single is defaulted to
`True`
```
##########
providers/google/src/airflow/providers/google/cloud/triggers/kubernetes_engine.py:
##########
@@ -274,7 +275,19 @@ def __init__(
self.ssl_ca_cert = ssl_ca_cert
self.job_name = job_name
self.job_namespace = job_namespace
- self.pod_name = pod_name
+ if pod_name is not None:
+ warnings.warn(
+ "`pod_name` parameter is deprecated and will be removed after
December 2025, "
+ "please use `pod_names` instead",
+ AirflowProviderDeprecationWarning,
+ stacklevel=2,
+ )
+ self.pod_name = pod_name
Review Comment:
Similar comment to CNCF's trigger
--
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]
