dabla commented on code in PR #58684:
URL: https://github.com/apache/airflow/pull/58684#discussion_r2563600240
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -851,32 +851,49 @@ def convert_config_file_to_dict(self):
else:
self._config_dict = None
- def invoke_defer_method(self, last_log_time: DateTime | None = None) ->
None:
+ def invoke_defer_method(self, context: Context, last_log_time: DateTime |
None = None) -> None:
"""Redefine triggers which are being used in child classes."""
self.convert_config_file_to_dict()
trigger_start_time = datetime.datetime.now(tz=datetime.timezone.utc)
- self.defer(
- trigger=KubernetesPodTrigger(
- pod_name=self.pod.metadata.name, # type: ignore[union-attr]
- pod_namespace=self.pod.metadata.namespace, # type:
ignore[union-attr]
- trigger_start_time=trigger_start_time,
- kubernetes_conn_id=self.kubernetes_conn_id,
- cluster_context=self.cluster_context,
- config_dict=self._config_dict,
- in_cluster=self.in_cluster,
- poll_interval=self.poll_interval,
- get_logs=self.get_logs,
- startup_timeout=self.startup_timeout_seconds,
- startup_check_interval=self.startup_check_interval_seconds,
- schedule_timeout=self.schedule_timeout_seconds,
- base_container_name=self.base_container_name,
- on_finish_action=self.on_finish_action.value,
- last_log_time=last_log_time,
- logging_interval=self.logging_interval,
- trigger_kwargs=self.trigger_kwargs,
- ),
- method_name="trigger_reentry",
+
+ trigger = KubernetesPodTrigger(
+ pod_name=self.pod.metadata.name, # type: ignore[union-attr]
+ pod_namespace=self.pod.metadata.namespace, # type:
ignore[union-attr]
+ trigger_start_time=trigger_start_time,
+ kubernetes_conn_id=self.kubernetes_conn_id,
+ cluster_context=self.cluster_context,
+ config_dict=self._config_dict,
+ in_cluster=self.in_cluster,
+ poll_interval=self.poll_interval,
+ get_logs=self.get_logs,
+ startup_timeout=self.startup_timeout_seconds,
+ startup_check_interval=self.startup_check_interval_seconds,
+ schedule_timeout=self.schedule_timeout_seconds,
+ base_container_name=self.base_container_name,
+ on_finish_action=self.on_finish_action.value,
+ last_log_time=last_log_time,
+ logging_interval=self.logging_interval,
+ trigger_kwargs=self.trigger_kwargs,
)
+ container_state = trigger.define_container_state(self.pod)
+ if container_state == ContainerState.TERMINATED or container_state ==
ContainerState.FAILED:
+ self.log.info("Skipping deferral as pod is already in a terminal
state")
+ self.trigger_reentry(
+ context=context,
+ event={
+ "status": "success" if container_state ==
ContainerState.TERMINATED else "failed",
+ "namespace": self.pod.metadata.namespace,
+ "name": self.pod.metadata.name,
+ "message": "Container succeeded"
+ if container_state == ContainerState.TERMINATED
+ else "Container failed",
+ "last_log_time": last_log_time,
+ **(self.trigger_kwargs or {}),
+ },
+ )
+ return
Review Comment:
I would write this as an if else statement instead of return, because you
actually don't return anything in this case, but that's a matter of taste,
don't now what the consensus is in this case for Airflow code base.
--
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]
