### Apache Airflow version

main (development)

### What happened

This is a random issue. The KubernetesPodOperator duplicates logs when they are interrupted.

### What you think should happen instead

When the logs are interrupted, the KubernetesPodOperator should continue logging since the last captured timestamp, and not re-print logs that were already printed.

### How to reproduce

If you run the following dag mulltiple times, you might get duplicated logs. Note that the issue is random and not easily reproducible:

```
import datetime

from airflow import models
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator

YESTERDAY = datetime.datetime.now() - datetime.timedelta(days=1)

with models.DAG(
    dag_id="composer_sample_kubernetes_pod",
    schedule_interval=datetime.timedelta(days=1),
    start_date=YESTERDAY,
) as dag:
    timeout = 240
    iterations = 1000
    arguments = \
        'for i in {1..%(iterations)s}; do echo "$i of %(iterations)s"; done' % {'iterations': iterations}

    kubernetes_min_pod_0 = KubernetesPodOperator(
        task_id="pod-ex-minimum-0",
        name="pod-ex-minimum-0",
        cmds=["/bin/bash", "-c"],
        arguments=[arguments],
        namespace="default",
        image="gcr.io/gcp-runtimes/ubuntu_18_0_4",
        startup_timeout_seconds=timeout
    )
```

Here is a sample output of duplicated logs:

[2023-08-18, 13:32:54 UTC] {kublogduplication.py:81} INFO - [base] 994 of 1000
[2023-08-18, 13:32:54 UTC] {kublogduplication.py:81} INFO - [base] 995 of 1000
[2023-08-18, 13:32:54 UTC] {kublogduplication.py:81} INFO - [base] 996 of 1000
[2023-08-18, 13:32:54 UTC] {kublogduplication.py:81} INFO - [base] 997 of 1000
[2023-08-18, 13:32:54 UTC] {kublogduplication.py:81} INFO - [base] 998 of 1000
[2023-08-18, 13:32:54 UTC] {kublogduplication.py:81} INFO - [base] 999 of 1000
[2023-08-18, 13:32:54 UTC] {kublogduplication.py:81} INFO - [base] 1000 of 1000
[2023-08-18, 13:32:54 UTC] {pod_manager.py:450} ERROR - Error parsing timestamp (no timestamp in message ''). Will continue execution but won't update timestamp
[2023-08-18, 13:32:54 UTC] {kublogduplication.py:81} INFO - [base] 
[2023-08-18, 13:32:54 UTC] {kublogduplication.py:116} WARNING - Pod pod-ex-minimum-0-b3bf545v log read interrupted but container base still running
[2023-08-18, 13:32:55 UTC] {before.py:35} INFO - Starting call to 'unusual_prefix_01add01418c7cdaf7afbb9c57d62a05595a85aa7_kublogduplication.CustomPodManager.fetch_container_logs.<locals>.consume_logs', this is the 1st time calling it.
[2023-08-18, 13:32:55 UTC] {kublogduplication.py:81} INFO - [base] 1 of 1000
[2023-08-18, 13:32:55 UTC] {kublogduplication.py:81} INFO - [base] 2 of 1000
[2023-08-18, 13:32:55 UTC] {kublogduplication.py:81} INFO - [base] 3 of 1000
[2023-08-18, 13:32:55 UTC] {kublogduplication.py:81} INFO - [base] 4 of 1000
[2023-08-18, 13:32:55 UTC] {kublogduplication.py:81} INFO - [base] 5 of 1000
[2023-08-18, 13:32:55 UTC] {kublogduplication.py:81} INFO - [base] 6 of 1000
[2023-08-18, 13:32:55 UTC] {kublogduplication.py:81} INFO - [base] 7 of 1000
[2023-08-18, 13:32:55 UTC] {kublogduplication.py:81} INFO - [base] 8 of 1000

### Operating System

Ubuntu

### Versions of Apache Airflow Providers

main

### Deployment

Google Cloud Composer

### Deployment details

_No response_

### Anything else

_No response_

### Are you willing to submit PR?

- [X] 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)

#
#
from __future__ import annotations
import logging
from datetime import datetime
from json.decoder import JSONDecodeError
from types import SimpleNamespace
from unittest import mock
from unittest.mock import MagicMock
import pendulum
import pytest
import time_machine
from kubernetes.client.rest import ApiException
from pendulum import DateTime
from pendulum.tz.timezone import Timezone
from urllib3.exceptions import HTTPError as BaseHTTPError
from airflow.exceptions import AirflowException
from airflow.providers.cncf.kubernetes.utils.pod_manager import (
    PodLogsConsumer,
    PodManager,
    PodPhase,
    container_is_running,
    container_is_succeeded,
    container_is_terminated,
)
from airflow.utils.timezone import utc
class TestPodManager: