### Description

Add a process_line callback to KubernetesPodOperator 
Like https://github.com/apache/airflow/blob/main/airflow/providers/apache/beam/operators/beam.py#LL304C36-L304C57 the `BeamRunPythonPipelineOperator`, which allows the user to add stateful plugins based on the logging from docker job 

### Use case/motivation

I can add a plugin based on the logging and also allow cleanup on_kill based on the job creation log. 

### Related issues

_No response_

### Are you willing to submit a PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

PodPhase,
    container_is_succeeded,
    get_container_termination_message,
)
from airflow.settings import pod_mutation_hook
from airflow.utils import yaml
from airflow.utils.helpers import prune_dict, validate_key
from airflow.utils.timezone import utcnow
from airflow.version import version as airflow_version
if TYPE_CHECKING:
    import jinja2
    from typing_extensions import Literal
    from airflow.providers.cncf.kubernetes.secret import Secret
    from airflow.utils.context import Context
alphanum_lower = string.ascii_lowercase + string.digits
KUBE_CONFIG_ENV_VAR = "KUBECONFIG"
def _rand_str(num):
    """Generate random lowercase alphanumeric string of length num.
    TODO: when min airflow version >= 2.5, delete this function and import from kubernetes_helper_functions.
    :meta private:
    """
    return "".join(secrets.choice(alphanum_lower) for _ in range(num))
def _add_pod_suffix(*, pod_name, rand_len=8, max_len=253):
    """Add random string to pod name while staying under max len.
    TODO: when min airflow version >= 2.5, delete this function and import from kubernetes_helper_functions.
    :meta private:
    """
    suffix = "-" + _rand_str(rand_len)
    return pod_name[: max_len - len(suffix)].strip("-.") + suffix
def _create_pod_id(