Prevent deferrable KubernetesPodOperator log parsing from blocking the triggerer event loop

[aeroyorch]

Prevent deferrable KubernetesPodOperator log parsing from blocking the triggerer event loop when default_deferrable is True with get_logs and logging_interval set, which stales on verbose pods.
Offload that work to a thread so the loop stays responsive.

---

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

• Yes (please specify the tool below)

Claude Opus 4.8 for tests.

---

• 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.

[aeroyorch]

I believe the CI errors are unrelated to the current changes.

[potiuk]

Thanks — this is the right concern. The triggerer multiplexes many triggers onto a single event loop, so any CPU-bound work in an async def stalls every other trigger, not just this one. A large decode() + splitlines(), and the per-line parse/emit loop, are both real candidates.

The restructuring is clean: _split_log_bytes and _emit_container_logs extracted so they can be handed to to_thread, with return now correctly relocated ahead of the extraction so the function's contract is unchanged.

I checked the thing that usually bites when you move logging into a worker thread — contextvars. asyncio.to_thread copies the current context (contextvars.copy_context()), so the self.log.log(...) calls inside _emit_container_logs keep whatever task/dag context is bound. No log-attribution loss.

One question worth a thought rather than a change: to_thread isn't free — each call dispatches to the executor and round-trips through the loop. For a pod emitting a handful of log lines per poll, that overhead may exceed the decode it's avoiding, and the triggerer polls frequently across many pods. Did you look at whether it's worth gating on payload size (offload only above some threshold), or is the constant cost small enough not to matter at the polling rates involved? Not blocking — just that "always offload" optimises the large-log case at some cost to the common small-log one.

---

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