### Apache Airflow version

2.6.0

### What happened

Stream logging works as long as the task is running. Once the task finishes, no logs are printed to the UI (only the hostname of the worker is printed)
<img width="1657" alt="image" src="https://user-images.githubusercontent.com/16529101/236212701-aecf6cdc-4d87-4817-a685-0778b94d182b.png">


### What you think should happen instead

Expected to see the complete log of a task

### How to reproduce

Start an airflow task. You should be able to see the logs coming in as a stream, once it finishes, the logs are gone

### Operating System

CentOS 7

### Versions of Apache Airflow Providers

airflow-provider-great-expectations==0.1.5
apache-airflow==2.6.0
apache-airflow-providers-airbyte==3.1.0
apache-airflow-providers-apache-hive==4.0.1
apache-airflow-providers-apache-spark==3.0.0
apache-airflow-providers-celery==3.1.0
apache-airflow-providers-common-sql==1.4.0
apache-airflow-providers-ftp==3.3.1
apache-airflow-providers-http==4.3.0
apache-airflow-providers-imap==3.1.1
apache-airflow-providers-mysql==5.0.0
apache-airflow-providers-oracle==3.4.0
apache-airflow-providers-postgres==5.4.0
apache-airflow-providers-sqlite==3.3.2

### Deployment

Virtualenv installation

### Deployment details

Celery with 4 workers nodes/VMs. Scheduler and Webserver on a different VM

### Anything else

_No response_

### Are you willing to submit 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)

ti = copy.copy(self.ti)
        ti.state = TaskInstanceState.SUCCESS
        
        with mock.patch("airflow.providers.amazon.aws.log.s3_task_handler.hasattr", return_value=False):
            log, metadata = self.s3_task_handler.read(ti, metadata={"log_pos": 0})
        assert 1 == len(log)
        assert log[0][0][-1].startswith("*** Falling back to local log")
    def test_read_when_s3_log_missing_and_log_pos_over_zero_pre_26(self):
        ti = copy.copy(self.ti)
        ti.state = TaskInstanceState.SUCCESS
        
        with mock.patch("airflow.providers.amazon.aws.log.s3_task_handler.hasattr", return_value=False):
            log, metadata = self.s3_task_handler.read(ti, metadata={"log_pos": 1})
        assert 1 == len(log)
        assert not log[0][0][-1].startswith("*** Falling back to local log")
    def test_s3_read_when_log_missing(self):
        handler = self.s3_task_handler
        url = "s3://bucket/foo"
        with mock.patch.object(handler.log, "error") as mock_error:
            result = handler.s3_read(url, return_error=True)
            msg = (
                f"Could not read logs from {url} with error: An error occurred (404) when calling the "
                f"HeadObject operation: Not Found"
            )
            assert result == msg
            mock_error.assert_called_once_with(msg, exc_info=True)
    def test_read_raises_return_error(self):