Fix 'do_xcom_push' and 'get_logs' functionality for KubernetesJobOperator

[MaksYermak]

In this PR I have fixed usage for do_xcom_push and get_logs functionality for KubernetesJobOperator in deferrable and non-deferrable modes.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[VladaZakharova]

Hi @romsharon98 and @amoghrajesh ! can you please check changes here? Thanks!

[VladaZakharova]

Hi @romsharon98, @amoghrajesh @potiuk @eladkal @hussein-awala !
Can you please check changes here?

[ccardi]

Not working on GKE Autopilot. Need to set get_logs=False.
Otherwise: pod_name is not defined.

---

ERROR - Task failed with exception
Traceback (most recent call last):
File "/opt/python3.11/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 465, in _execute_task
result = _execute_callable(context=context, **execute_callable_kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/python3.11/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 432, in _execute_callable
return execute_callable(context=context, **execute_callable_kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/python3.11/lib/python3.11/site-packages/airflow/models/baseoperator.py", line 1724, in resume_execution
return execute_callable(context)
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/gcs/dags/dag_gke_pod-simple-pi-joboperator.py", line 30, in execute_complete
super().execute_complete(context, event)
File "/opt/python3.11/lib/python3.11/site-packages/airflow/providers/cncf/kubernetes/operators/job.py", line 233, in execute_complete
pod_name = event["pod_name"]

KeyError: 'pod_name'

[NilsIrl]

How does this work if the pod wasn't already created by the time this is reached? Won't 2 pods eventually get created (when the job eventually creates the pod)?

[steinwaywhw]

I have the same question. I thought the K8s Job will manage the pod creations, why would we need creating the pod manually again? I'm actually running into bugs where I set parallelism to more than 1 and this line triggers self.find_pods() which triggers

airflow/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py

Line 563 in 56fbe90

raise AirflowException(f"More than one pod running with labels {label_selector}")

[amoghrajesh]

@steinwaywhw can you explain your problem more? Would you like to take a stab at fixing it?

[MaksYermak]

Hello @steinwaywhw,
It's a known issue #44994 and the fix was already prepared by this PR #49899