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

image_pull_secrets=self.image_pull_secrets,
                service_account_name=self.service_account_name,
                host_network=self.hostnetwork,
                hostname=self.hostname,
                subdomain=self.subdomain,
                security_context=self.security_context,
                dns_policy=self.dnspolicy,
                dns_config=self.dns_config,
                scheduler_name=self.schedulername,
                restart_policy="Never",
                priority_class_name=self.priority_class_name,
                volumes=self.volumes,
                active_deadline_seconds=self.active_deadline_seconds,
            ),
        )
        pod = PodGenerator.reconcile_pods(pod_template, pod)
        if not pod.metadata.name:
            pod.metadata.name = _create_pod_id(
                task_id=self.task_id, unique=self.random_name_suffix, max_length=80
            )
        elif self.random_name_suffix:
            
            pod.metadata.name = _add_pod_suffix(pod_name=pod.metadata.name)
        if not pod.metadata.namespace:
            hook_namespace = self.hook.get_namespace()
            pod_namespace = self.namespace or hook_namespace or self._incluster_namespace or "default"
            pod.metadata.namespace = pod_namespace
        for secret in self.secrets:
            self.log.debug("Adding secret to task %s", self.task_id)
            pod = secret.attach_to_pod(pod)