Fix duplicate pod creation in KubernetesJobOperator

[stephen-bracken]

Fix KubernetesJobOperator.get_or_create_pod() sometimes creating duplicate pods.

(re-raised from #52885)

during execute() the KubernetesJobOperator attempts to find the pod associated with the Job object using self.get_or_create_pod(). If Kubernetes is being slow then the Job object will not create a pod before this method gets called, which will result in the underlying find_pod() method returning None, and a duplicate headless pod being created for this task.

This PR removes references to the get_or_create_pod() method in favour of KubernetesJobOperator.get_pod() to prevent creating headless pods.

---

^ 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 airflow-core/newsfragments.

[shahar1]

Overall LGTM - got some minor comments.
Also, I'll be happy for an additional review :)

[stephen-bracken]

In reference to #49899 - I think a change is still necessary to avoid creating multiple pods when parallelism is not set. Do we want to change the control flow to use get_pods() in all circumstances?

[shahar1]

In reference to #49899 - I think a change is still necessary to avoid creating multiple pods when parallelism is not set. Do we want to change the control flow to use get_pods() in all circumstances?

You could start with handling only the case where parallelism=False, and later we could simplify the logic if it becomes necessary. Please rebase/merge changes from the main branch, adjust the logic and fix tests appropriately.

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.

[shahar1]

LGTM, I'm making sure with the other contributors that handling parallelism = 0 case makes sense:
https://apache-airflow.slack.com/archives/C06K9Q5G2UA/p1767429130216899

If no strong objections are given in the next 1-2 days, or there are additional approvals for this PR by then - I'm ok with merging it as-is.

[jscheffl]

@stephen-bracken / @shahar1 as concerns raised in https://apache-airflow.slack.com/archives/C06K9Q5G2UA/p1767429130216899 - one thing to consider maybe is - I'd propose to - adding a newsfragment such that it is highlighted in the changelog of the provider.

TLDR: Requesting to add a newsfragment to highlight this interface change.

UPDATE: Args, providers have no newsfragmen, add it to changelogs like in https://github.com/apache/airflow/pull/59143/changes#diff-24cff4e7b7926b95f4efef45da9f9d6f43b237b5143990b1554113251cb2c12eR30 for example that it is included in next providers wave.

[stephen-bracken]

@stephen-bracken / @shahar1 as concerns raised in https://apache-airflow.slack.com/archives/C06K9Q5G2UA/p1767429130216899 - one thing to consider maybe is - I'd propose to - adding a newsfragment such that it is highlighted in the changelog of the provider.

TLDR: Requesting to add a newsfragment to highlight this interface change.

UPDATE: Args, providers have no newsfragmen, add it to changelogs like in https://github.com/apache/airflow/pull/59143/changes#diff-24cff4e7b7926b95f4efef45da9f9d6f43b237b5143990b1554113251cb2c12eR30 for example that it is included in next providers wave.

@jscheffl Thanks, added a changelog note.

[shahar1]

Drafting the PR following the author's request to make some changes, please do not merge

[shahar1]

See previous comment

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.

[shahar1]

Great job @stephen-bracken !