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

return str_tpl.format(
                dag_id=ti.dag_id,
                task_id=ti.task_id,
                run_id=ti.run_id,
                data_interval_start=data_interval_start,
                data_interval_end=data_interval_end,
                execution_date=ti.get_dagrun().logical_date.isoformat(),
                try_number=try_number,
            )
        else:
            raise RuntimeError(f"Unable to render log filename for {ti}. This should never happen")
    def _read_grouped_logs(self):
        return False
    @cached_property
    def _executor_get_task_log(self) -> Callable[[TaskInstance, int], tuple[list[str], list[str]]]:
        """This cached property avoids loading executor repeatedly."""
        executor = ExecutorLoader.get_default_executor()
        return executor.get_task_log
    def _read(
        self,
        ti: TaskInstance,
        try_number: int,
        metadata: dict[str, Any] | None = None,
    ):
        """
        Template method that contains custom logic of reading
        logs given the try_number.
        :param ti: task instance record
        :param try_number: current try_number to read log from
        :param metadata: log metadata,