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

self.remote_pod = self.find_pod(self.pod.metadata.namespace, context=context)
            self.await_pod_start(pod=self.pod)
            if self.get_logs:
                self.pod_manager.fetch_requested_container_logs(
                    pod=self.pod,
                    container_logs=self.container_logs,
                    follow_logs=True,
                )
            if not self.get_logs or (
                self.container_logs is not True and self.base_container_name not in self.container_logs
            ):
                self.pod_manager.await_container_completion(
                    pod=self.pod, container_name=self.base_container_name
                )
            if self.do_xcom_push:
                self.pod_manager.await_xcom_sidecar_container_start(pod=self.pod)
                result = self.extract_xcom(pod=self.pod)
            istio_enabled = self.is_istio_enabled(self.pod)
            self.remote_pod = self.pod_manager.await_pod_completion(
                self.pod, istio_enabled, self.base_container_name
            )
        finally:
            self.cleanup(
                pod=self.pod or self.pod_request_obj,
                remote_pod=self.remote_pod,
            )
        if self.do_xcom_push:
            return result
    def execute_async(self, context: Context):