Add opt-in concurrent pod creation to KubernetesExecutor

[1fanwang]

Why

KubernetesExecutor creates worker pods one at a time — each create call blocks before the next
starts. At production create latencies (~150–500ms of round-trip plus admission webhooks), the
executor, not the cluster, caps task startup. The existing batch size dequeues more pods per loop
but still creates them serially, so per-create latency re-serializes them.

The cost lands on task.queued_duration: it grows with a task's position in the batch, so a
fan-out's average and p99 inflate while the first task is unaffected. Dynamic task mapping makes
large fan-outs routine.

What

Opt-in, off by default. The pods dequeued in a scheduler loop are created concurrently instead of
serially, bounded by a configurable in-flight limit. The serial path stays the default and
unchanged (retry, exceeded-quota, rate-limit backoff). A new metric reports per-loop batch
creation time on both paths.

Benchmark

Live multi-node kind, per-create latency injected by an admission webhook.

Create serialization (webhook admission timestamps, which isolate the create call from pod start),
30-pod batch, concurrency 32:

create component	serial	async
average @400ms	6.3s	0.2s
p99 @400ms	12.3s	1.1s

Tail (last task) across the production band — serial scales with latency and position, async stays flat:

per-create latency	serial	async
300ms	9.4s	1.1s
400ms	12.4s	1.1s
500ms	15.2s	1.1s

End-to-end task.queued_duration, read from the metadata DB (start_date − queued_dttm) per mapped
task — the actual metric, not derived. 30-way fan-out; per-create latency raised to 2s so the create
serialization dominates the single-host kubelet-start floor (a small-cluster artifact — a real
multi-node cluster starts the batch in parallel):

task.queued_duration	serial	async
average	34.7s	18.2s
p95	60.7s	19.9s
p99	63.0s	20.1s
tail	63.5s	20.2s

Average falls 1.9×, p99 and tail 3.1×. It is opt-in because at low latency or small batches the gap
shrinks toward zero and async carries a small fixed overhead.

Verification

Unit: serial path unchanged; new coverage for concurrent creation, the concurrency bound, error
handling and backoff on the concurrent path, per-pod failure isolation, and the Airflow 3 path.
E2E: live kind, real apiserver — the concurrent path creates pods as measured above.

Hardening (from review)

_handle_pod_publish_error now re-queues transient failures it previously dropped, on both paths:
502/503/504 (503's Retry-After honored like 429) and connection-level errors on the sync
(urllib3) and async (aiohttp) clients — the latter previously crashed the sequential loop.

E2E — error injected on the first CREATE per task, task_publish_max_retries=3

inject on 1st CREATE	path	before	after
503 + Retry-After	async, sync	task fails	6 / 6 pods, 0 failed
connection drop	async (aiohttp)	task fails (0 / 6)	6 / 6 pods, 0 failed
connection drop	sync (urllib3)	sync loop crashes	6 / 6 pods, 0 failed

>>> ASYNC 503:            pods=6 failed_events=0
>>> SYNC  503:            pods=6 failed_events=0
>>> ASYNC conn (aiohttp): pods=6 failed_events=0
>>> SYNC  conn (urllib3): pods=6 failed_events=0
>>> BEFORE ASYNC conn:    pods=0 failed_events=6   # base, no hardening

Open question: a connection error can arrive after the pod was created (reply lost), so re-queue can
double-create — same risk as today's 500 re-queue. Gate to connect-phase failures?

[jscheffl]

I am not a KubernetesExecutor expert as not using it but as I am an asyncio fanboy I really understand and like the improvement.

I hope that this feature matures over time and then the sync and async code streams can be consolidated (meaning: drop the sync option because in the way implemented and K8s API client restricts it needs two parallel implementation streams to maintain).

As not being expert requesting another pair of eyes as review, my comments are just from reading not from running myself.

[shahar1]

Please resolve conflicts

[potiuk]

@1fanwang — this PR has 1 unresolved review thread(s) that still look like they need your attention. Once you've addressed them (push changes and/or reply in-thread), please resolve the threads or reply to confirm, and give the reviewer a nudge for another look. Thanks!

See the PR quality criteria.

_{Automated first-pass triage note drafted by an AI-assisted tool — may get things wrong; once addressed, a real Apache Airflow maintainer takes the next look. (why automated)}

---

Drafted-by: Claude Code (Opus 4.8); reviewed by @potiuk before posting

[1fanwang]

@1fanwang — this PR has 1 unresolved review thread(s) that still look like they need your attention. Once you've addressed them (push changes and/or reply in-thread), please resolve the threads or reply to confirm, and give the reviewer a nudge for another look. Thanks!

See the PR quality criteria.

Automated first-pass triage note drafted by an AI-assisted tool — may get things wrong; once addressed, a real Apache Airflow maintainer takes the next look. (why automated)

Drafted-by: Claude Code (Opus 4.8); reviewed by @potiuk before posting

thanks for the ping, resolved

[shahar1]

@1fanwang Could you please resolve conflicts?

[shahar1]

@1fanwang could you please resolve conflicts again?