### 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)

"""
        if not context:
            return {}
        ti = context["ti"]
        run_id = context["run_id"]
        labels = {
            "dag_id": ti.dag_id,
            "task_id": ti.task_id,
            "run_id": run_id,
            "kubernetes_pod_operator": "True",
        }
        map_index = ti.map_index
        if map_index >= 0:
            labels["map_index"] = map_index
        if include_try_number:
            labels.update(try_number=ti.try_number)
        
        if context["dag"].parent_dag:
            labels["parent_dag_id"] = context["dag"].parent_dag.dag_id
        
        
        for label_id, label in labels.items():
            safe_label = pod_generator.make_safe_label_value(str(label))
            labels[label_id] = safe_label
        return labels
    @cached_property
    def pod_manager(self) -> PodManager:
        return PodManager(kube_client=self.client)
    @cached_property
    def hook(self) -> PodOperatorHookProtocol: