Kubernetes Pod Operator - handle pod preemption before container creation

[johnhoran]

We have been seeing occasional failures on KPO where a pod gets preempted and then hangs, usually until the dag hits its dagrun timeout, or just for an indeterminate amount of time, once execution continues it logs that the pod doesn't exist.

What I believe is happening is that the pod is getting preempted before any containers are created, and the pod gets assigned a failed phase, however the container_statuses array is empty. Looking at my logs I get a kubernetes manifest with a status like

'status': {
            'container_statuses': None,
            'init_container_statuses': None,
            'message': "Pod was rejected: Node didn't have enough resource: "
                       'memory, requested: 547356672, used: 6619660288, '
                       'capacity: 7068438528',
            'phase': 'Failed',
            'reason': 'OutOfmemory',
            'start_time': datetime.datetime(2026, 5, 27, 15, 33, 43, tzinfo=tzlocal()),
...
}

After implementing this I noticed #66650 which I think probably fixes the issue the majority of the time, but my only misgiving there is that I noticed in my logs instances of the kubernetes manifest being printed after the failure with a status like

'status': {'allocated_resources': None,
            'conditions': [{'last_probe_time': None,
                            'last_transition_time': datetime.datetime(2026, 5, 14, 0, 2, 25, tzinfo=tzlocal()),
                            'message': '0/18 nodes are available: 2 node(s) '
                                       "didn't match Pod's node "
                                       'affinity/selector, 2 node(s) had '
                                       'untolerated taint '
                                       '{ebs.csi.aws.com/agent-not-ready: }, 2 '
                                       'node(s) had untolerated taint '
                                       '{eks.amazonaws.com/compute-type: '
                                       'fargate}, 4 node(s) had untolerated '
                                       'taint {astronomer.io/node-group: '
                                       'airflow-system}, 8 Insufficient '
                                       'memory. preemption: not eligible due '
                                       'to preemptionPolicy=Never.',
                            'observed_generation': None,
                            'reason': 'Unschedulable',
                            'status': 'False',
                            'type': 'PodScheduled'}],
            'container_statuses': None,
            'init_container_statuses': None,
            'message': None,
            'phase': 'Pending',
            'start_time': None
...
}

Which suggests that the pod transitioned from failed back to pending. This is from a seperate execution, and its beyond me to reproduce it to allow me to understand exactly what kubernetes is doing. But since the call to define_pod_container_state involves another API call, I'm concerned it might have already transitioned back to pending by the time it is sent to that method.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.