Fix monitoring-pod leak in KubernetesJobOperator (#67333)

* Fixing pod leak in KubernetesJobOperator (#1)

* fix(providers/cncf/kubernetes): clean up monitoring pods in KubernetesJobOperator

KubernetesJobOperator inherited from KubernetesPodOperator but overrode
execute() without calling post_complete_action(), so the monitoring /
log-streaming pods discovered via get_pods() were never deleted. These
pods have no ownerReferences to the V1Job, so ttl_seconds_after_finished
and the Foreground cascade in on_kill don't reap them either.

- execute() and execute_complete() now wrap their work in try/finally and
  call post_complete_action() for each pod in self.pods. on_finish_action
  (delete_pod / delete_succeeded_pod / keep_pod) is now honoured.
- on_kill() additionally calls pod_manager.delete_pod() for each
  monitoring pod (the Job's foreground cascade doesn't reach them).
- Per-pod cleanup errors are logged but never mask the in-flight
  exception, so Job-level failures keep propagating.
- execute_complete() resolves monitoring pods once and shares the lookup
  between the log-retrieval path and the cleanup path.
- Added unit tests, a bugfix newsfragment, and an operators.rst section
  documenting the cleanup contract.

* Address code review feedback: remove dead PodNotFoundException check, drop unused import, relax pod-deletion ordering in test, fix trailing comma

* Potential fix for pull request finding

In _cleanup_monitoring_pods, remote_pod is resolved via find_pod(), which is designed to locate a single matching pod by task-instance labels and can invoke duplicate-pod resolution logic (process_duplicate_label_pods). For KubernetesJobOperator with parallelism > 1, this lookup can return the wrong pod (or trigger duplicate-handling side effects), so post_complete_action() may receive a mismatched remote_pod. Consider using the already-discovered pod’s name/namespace to refresh state (e.g. via hook.get_pod) or just pass remote_pod=pod when you already have the V1Pod object from get_pods().

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Use isinstance(exc, TaskDeferred) instead of brittle string comparison

* Potential fix for pull request finding

The new unit tests add several mock.MagicMock() instances (pods, jobs, TI, etc.) without spec/autospec, and some patch() usages also create non-spec'd mocks by default. Using autospec=True on patches and create_autospec(...)/MagicMock(spec=...) for key Kubernetes objects helps catch typos/attribute mismatches in these tests and aligns with Airflow’s test hardening guidance.

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Address PR review comments: fix trigger pod_names, on_kill logging, and test assertions

- triggers/job.py: Always include pod_names/pod_namespace in trigger event
  regardless of get_logs setting, so execute_complete() can reliably clean
  up monitoring pods even when get_logs=False
- operators/job.py: Log unexpected ApiException in on_kill() instead of
  suppressing all ApiExceptions; remove unused `suppress` import
- tests/test_job.py: Rewrite test_execute_respects_keep_pod and
  test_execute_deletes_pod_default to keep process_pod_deletion real and
  assert on pod_manager.delete_pod; stub hook.get_pod for remote_pod resolution
- tests/test_job.py: Add regression test for get_logs=False deferrable path

* Fix orphaned test_on_kill_deletes_monitoring_pods method body after accidental deletion of method signature

* Make pod resolution best-effort in execute_complete

* Address remaining KubernetesJobOperator review comments

* Finalize review-comment fixes for KubernetesJobOperator

* Fix remaining KubernetesJobOperator review comments

* Update KubernetesJobOperator docs for action semantics

* Improve KubernetesJobOperator newsfragment readability

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: Ville Jyrkkä <vjyrkka@gmail.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Apply ruff format

* Move provider changelog entry; drop newsfragment

* Apply prek docstring auto-fix (D212 multi-line summary)

* Drop changelog entry — added by release manager at release time

* Cover remaining KubernetesJobOperator monitoring-pod cleanup paths

* Consolidate PR tests: drop call-counting, dedupe, observe state

* Apply test review fixes: hoist imports, add state assertions, dedupe with _pod helper

* Fix Copilot review: spec all MagicMock usages in KubernetesJobOperator tests

- _recording_pod_manager(): use create_autospec(PodManager) instead of MagicMock
- test error masking: use create_autospec(PodManager) for boom-raising pm
- test_execute_does_not_cleanup_when_deferring: use create_autospec(KubernetesJobTrigger) for trigger
- test_execute_complete_pod_api_error_does_not_raise: use plain {} for event["job"] field
- test_run_success_emits_pod_info_when_get_logs_false (triggers): use create_autospec(k8s.V1Job)

---------

Co-authored-by: Copilot <198982749+Copilot@users.noreply.github.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>