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

as the impersonated user. However, if the user manually runs tasks
        of the SubDagOperator through the UI, then the log files are created
        by the user that runs the Airflow command. For example, the Airflow
        run command may be run by the `airflow_sudoable` user, but the Airflow
        tasks may be run by the `airflow` user. If the log files are not
        writable by both users, then it's possible that re-running a task
        via the UI (or vice versa) results in a permission error as the task
        tries to write to a log file created by the other user.
        We leave it up to the user to manage their permissions by exposing configuration for both
        new folders and new log files. Default is to make new log folders and files group-writeable
        to handle most common impersonation use cases. The requirement in this case will be to make
        sure that the same group is set as default group for both - impersonated user and main airflow
        user.
        """
        new_folder_permissions = int(
            conf.get("logging", "file_task_handler_new_folder_permissions", fallback="0o775"), 8
        )
        directory.mkdir(mode=new_folder_permissions, parents=True, exist_ok=True)
        if directory.stat().st_mode % 0o1000 != new_folder_permissions % 0o1000:
            print(f"Changing {directory} permission to {new_folder_permissions}")
            try:
                directory.chmod(new_folder_permissions)
            except PermissionError as e:
                
                
                
                
                print(f"Failed to change {directory} permission to {new_folder_permissions}: {e}")
                pass
    def _init_file(self, ti):