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

istio_enabled = self.is_istio_enabled(remote_pod)
        pod_phase = remote_pod.status.phase if hasattr(remote_pod, "status") else None
        
        if pod_phase != PodPhase.SUCCEEDED or self.on_finish_action == OnFinishAction.KEEP_POD:
            self.patch_already_checked(remote_pod, reraise=False)
        if (pod_phase != PodPhase.SUCCEEDED and not istio_enabled) or (
            istio_enabled and not container_is_succeeded(remote_pod, self.base_container_name)
        ):
            if self.log_events_on_failure:
                self._read_pod_events(pod, reraise=False)
            self.process_pod_deletion(remote_pod, reraise=False)
            error_message = get_container_termination_message(remote_pod, self.base_container_name)
            if self.skip_on_exit_code is not None:
                container_statuses = (
                    remote_pod.status.container_statuses if remote_pod and remote_pod.status else None
                ) or []
                base_container_status = next(
                    (x for x in container_statuses if x.name == self.base_container_name), None
                )
                exit_code = (
                    base_container_status.state.terminated.exit_code
                    if base_container_status
                    and base_container_status.state
                    and base_container_status.state.terminated
                    else None
                )
                if exit_code in self.skip_on_exit_code: