Kubernetes Pod Operator callbacks repeating log line

[johnhoran]

Closes #59366

I noticed the callbacks were just being passed the same log line over and over again. I tried previous versions of the provider, and they all seem to have the same bug, so I think this probably never worked correctly.

I think the best thing to do is have the callback be passed the same message that is passed to the logging call, and additionally to add some extra context, like the container name, timestamp and the pod object. Given that this feature hasn't been working before now, I hope it will be okay to modify how it functions like this. The motivation for passing the pod is that I'm playing around with another PR that sets xcoms in a callback attached in deffered mode, and the pod contains the context necessary to set the xcoms in its labels, and it is also available on the triggerer when running in deferred mode, so passing it can be supported there too.

I've also expanded the test to catch this bug in future.

The cause of the bug was the reuse of the line variable from

airflow/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/utils/pod_manager.py

Line 482 in 0d016a6

line = raw_line.decode("utf-8", errors="backslashreplace")

on

airflow/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/utils/pod_manager.py

Line 490 in 0d016a6

for line in progress_callback_lines:

which then gets put into the array after the iteration loop has complete

airflow/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/utils/pod_manager.py

Line 505 in 0d016a6

progress_callback_lines = [line]

so it ends up just carrying the same line and putting it back into iteration over and over again.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[tatiana]

Thanks a lot for the reviews, @ephraimbuddy and @vatsrahul1001 !

Just wanted to share that we changed the implementation in Cosmos to use the ones proposed by @johnhoran in an ongoing task astronomer/astronomer-cosmos#2207. We were able to reproduce the problem previously, and I can confirm that after the changes, the behaviour works as expected. I validated them both via integration tests and also by running Airflow and triggering a DAG manually.