### Body

Use stacklevel 2 instead of the default 1:
https://github.com/apache/airflow/blob/b5057e0e1fc6b7a47e38037a97cac862706747f0/airflow/providers/cncf/kubernetes/utils/pod_manager.py#L361-L365

### Committer

- [X] I acknowledge that I am a maintainer/committer of the Apache Airflow project.
"""
    Examines V1Pod ``pod`` to determine whether ``container_name`` is completed and succeeded.
    If that container is present and completed and succeeded, returns True.  Returns False otherwise.
    """
    if not container_is_completed(pod, container_name):
        return False
    container_status = get_container_status(pod, container_name)
    if not container_status:
        return False
    return container_status.state.terminated.exit_code == 0
def container_is_terminated(pod: V1Pod, container_name: str) -> bool:
    """
    Examines V1Pod ``pod`` to determine whether ``container_name`` is terminated.
    If that container is present and terminated, returns True.  Returns False otherwise.
    """
    container_statuses = pod.status.container_statuses if pod and pod.status else None
    if not container_statuses:
        return False
    container_status = next((x for x in container_statuses if x.name == container_name), None)
    if not container_status:
        return False
    return container_status.state.terminated is not None
def get_container_termination_message(pod: V1Pod, container_name: str):
    with suppress(AttributeError, TypeError):
        container_statuses = pod.status.container_statuses
        container_status = next((x for x in container_statuses if x.name == container_name), None)
        return container_status.state.terminated.message if container_status else None
class PodLogsConsumer: