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

[2023-01-15T22:37:04.351-0800] {temporal.py:71} INFO - sleeping 1 second...
[2023-01-15T22:37:05.353-0800] {temporal.py:71} INFO - sleeping 1 second...
[2023-01-15T22:37:06.354-0800] {temporal.py:71} INFO - sleeping 1 second...
[2023-01-15T22:37:07.355-0800] {temporal.py:71} INFO - sleeping 1 second...
[2023-01-15T22:37:08.356-0800] {temporal.py:71} INFO - sleeping 1 second...
[2023-01-15T22:37:09.357-0800] {temporal.py:71} INFO - sleeping 1 second...
[2023-01-15T22:37:10.358-0800] {temporal.py:71} INFO - sleeping 1 second...
[2023-01-15T22:37:11.359-0800] {temporal.py:71} INFO - sleeping 1 second...
[2023-01-15T22:37:12.359-0800] {temporal.py:71} INFO - sleeping 1 second...
[2023-01-15T22:37:13.360-0800] {temporal.py:74} INFO - yielding event with payload DateTime(2023, 1, 16, 6, 37, 13, 44492, tzinfo=Timezone('UTC'))
[2023-01-15T22:37:13.361-0800] {triggerer_job.py:540} INFO - Trigger <airflow.triggers.temporal.DateTimeTrigger moment=2023-01-16T06:37:13.044492+00:00> (ID 106) fired: TriggerEvent<DateTime(2023, 1, 16, 6, 37, 13, 44492, tzinfo=Timezone('UTC'))>
"""  
def test_this():
    """
    Notice there are two messages with timestamp `2023-01-17T12:47:11.883-0800`.
    In this case, these should appear in correct order and be deduped in result.
    """
    sample_with_dupe = """[2023-01-17T12:46:55.868-0800] {temporal.py:62} INFO - trigger starting
    [2023-01-17T12:46:55.868-0800] {temporal.py:71} INFO - sleeping 1 second...
    [2023-01-17T12:47:09.882-0800] {temporal.py:71} INFO - sleeping 1 second...
    [2023-01-17T12:47:10.882-0800] {temporal.py:71} INFO - sleeping 1 second...
    [2023-01-17T12:47:11.883-0800] {temporal.py:74} INFO - yielding event with payload DateTime(2023, 1, 17, 20, 47, 11, 254388, tzinfo=Timezone('UTC'))
    [2023-01-17T12:47:11.883-0800] {triggerer_job.py:540} INFO - Trigger <airflow.triggers.temporal.DateTimeTrigger moment=2023-01-17T20:47:11.254388+00:00> (ID 1) fired: TriggerEvent<DateTime(2023, 1, 17, 20, 47, 11, 254388, tzinfo=Timezone('UTC'))>
    """  
    assert sample_with_dupe == "\n".join(_interleave_logs(sample_with_dupe, "", sample_with_dupe))
def test_ttthis():
    local_logs = "[2023-01-17T13:41:53.228-0800] {temporal.py:62} INFO - trigger starting\n[2023-01-17T13:41:53.228-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:41:54.228-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:41:55.229-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:41:56.229-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:41:57.229-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:41:58.230-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:41:59.231-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:42:00.231-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:42:01.232-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:42:02.233-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:42:03.233-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:42:04.233-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:42:05.233-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:42:06.234-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:42:07.235-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:42:08.237-0800] {temporal.py:74} INFO - yielding event with payload DateTime(2023, 1, 17, 21, 42, 8, 150776, tzinfo=Timezone('UTC'))\n[2023-01-17T13:42:08.237-0800] {triggerer_job.py:540} INFO - Trigger <airflow.triggers.temporal.DateTimeTrigger moment=2023-01-17T21:42:08.150776+00:00> (ID 4) fired: TriggerEvent<DateTime(2023, 1, 17, 21, 42, 8, 150776, tzinfo=Timezone('UTC'))>\n[2023-01-17T13:42:19.050-0800] {taskinstance.py:1131} INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: example_time_delta_sensor_async.wait manual__2023-01-17T21:41:38.150776+00:00 [queued]>\n[2023-01-17T13:42:19.074-0800] {taskinstance.py:1131} INFO - Dependencies all met for dep_context=requeueable deps ti=<TaskInstance: example_time_delta_sensor_async.wait manual__2023-01-17T21:41:38.150776+00:00 [queued]>\n[2023-01-17T13:42:19.074-0800] {taskinstance.py:1330} INFO - resuming after deferral\n[2023-01-17T13:42:19.087-0800] {taskinstance.py:1351} INFO - Executing <Task(TimeDeltaSensorAsync): wait> on 2023-01-17 21:41:38.150776+00:00\n[2023-01-17T13:42:19.091-0800] {standard_task_runner.py:56} INFO - Started process 98604 to run task\n[2023-01-17T13:42:19.097-0800] {standard_task_runner.py:83} INFO - Running: ['airflow', 'tasks', 'run', 'example_time_delta_sensor_async', 'wait', 'manual__2023-01-17T21:41:38.150776+00:00', '--job-id', '518', '--raw', '--subdir', '/Users/dstandish/code/airflow/airflow/example_dags/example_time_delta_sensor_async.py', '--cfg-path', '/var/folders/7_/1xx0hqcs3txd7kqt0ngfdjth0000gn/T/tmpx76emkve', '--no-shut-down-logging']\n[2023-01-17T13:42:19.101-0800] {standard_task_runner.py:84} INFO - Job 518: Subtask wait\n[2023-01-17T13:42:19.185-0800] {task_command.py:417} INFO - Running <TaskInstance: example_time_delta_sensor_async.wait manual__2023-01-17T21:41:38.150776+00:00 [running]> on host Daniels-MacBook-Pro-2.local\n[2023-01-17T13:42:19.274-0800] {taskinstance.py:1369} INFO - Marking task as SUCCESS. dag_id=example_time_delta_sensor_async, task_id=wait, execution_date=20230117T214138, start_date=20230117T214151, end_date=20230117T214219\n[2023-01-17T13:42:19.309-0800] {local_task_job.py:220} INFO - Task exited with return code 0\n[2023-01-17T13:42:19.339-0800] {taskinstance.py:2648} INFO - 1 downstream tasks scheduled from follow-on schedule check\n"  
    remote_logs = "[2023-01-17T13:41:51.446-0800] {taskinstance.py:1131} INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: example_time_delta_sensor_async.wait manual__2023-01-17T21:41:38.150776+00:00 [queued]>\n[2023-01-17T13:41:51.454-0800] {taskinstance.py:1131} INFO - Dependencies all met for dep_context=requeueable deps ti=<TaskInstance: example_time_delta_sensor_async.wait manual__2023-01-17T21:41:38.150776+00:00 [queued]>\n[2023-01-17T13:41:51.454-0800] {taskinstance.py:1332} INFO - Starting attempt 1 of 1\n[2023-01-17T13:41:51.478-0800] {taskinstance.py:1351} INFO - Executing <Task(TimeDeltaSensorAsync): wait> on 2023-01-17 21:41:38.150776+00:00\n[2023-01-17T13:41:51.482-0800] {standard_task_runner.py:56} INFO - Started process 98256 to run task\n[2023-01-17T13:41:51.487-0800] {standard_task_runner.py:83} INFO - Running: ['airflow', 'tasks', 'run', 'example_time_delta_sensor_async', 'wait', 'manual__2023-01-17T21:41:38.150776+00:00', '--job-id', '517', '--raw', '--subdir', '/Users/dstandish/code/airflow/airflow/example_dags/example_time_delta_sensor_async.py', '--cfg-path', '/var/folders/7_/1xx0hqcs3txd7kqt0ngfdjth0000gn/T/tmpszoemoy8', '--no-shut-down-logging']\n[2023-01-17T13:41:51.490-0800] {standard_task_runner.py:84} INFO - Job 517: Subtask wait\n[2023-01-17T13:41:51.575-0800] {task_command.py:417} INFO - Running <TaskInstance: example_time_delta_sensor_async.wait manual__2023-01-17T21:41:38.150776+00:00 [running]> on host Daniels-MacBook-Pro-2.local\n[2023-01-17T13:41:51.651-0800] {taskinstance.py:1558} INFO - Exporting env vars: AIRFLOW_CTX_DAG_OWNER='airflow' AIRFLOW_CTX_DAG_ID='example_time_delta_sensor_async' AIRFLOW_CTX_TASK_ID='wait' AIRFLOW_CTX_EXECUTION_DATE='2023-01-17T21:41:38.150776+00:00' AIRFLOW_CTX_TRY_NUMBER='1' AIRFLOW_CTX_DAG_RUN_ID='manual__2023-01-17T21:41:38.150776+00:00'\n[2023-01-17T13:41:51.672-0800] {taskinstance.py:1433} INFO - Pausing task as DEFERRED. dag_id=example_time_delta_sensor_async, task_id=wait, execution_date=20230117T214138, start_date=20230117T214151\n[2023-01-17T13:41:51.700-0800] {local_task_job.py:218} INFO - Task exited with return code 100 (task deferral)\n\n[2023-01-17T13:42:19.050-0800] {taskinstance.py:1131} INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: example_time_delta_sensor_async.wait manual__2023-01-17T21:41:38.150776+00:00 [queued]>\n[2023-01-17T13:42:19.074-0800] {taskinstance.py:1131} INFO - Dependencies all met for dep_context=requeueable deps ti=<TaskInstance: example_time_delta_sensor_async.wait manual__2023-01-17T21:41:38.150776+00:00 [queued]>\n[2023-01-17T13:42:19.074-0800] {taskinstance.py:1330} INFO - resuming after deferral\n[2023-01-17T13:42:19.087-0800] {taskinstance.py:1351} INFO - Executing <Task(TimeDeltaSensorAsync): wait> on 2023-01-17 21:41:38.150776+00:00\n[2023-01-17T13:42:19.091-0800] {standard_task_runner.py:56} INFO - Started process 98604 to run task\n[2023-01-17T13:42:19.097-0800] {standard_task_runner.py:83} INFO - Running: ['airflow', 'tasks', 'run', 'example_time_delta_sensor_async', 'wait', 'manual__2023-01-17T21:41:38.150776+00:00', '--job-id', '518', '--raw', '--subdir', '/Users/dstandish/code/airflow/airflow/example_dags/example_time_delta_sensor_async.py', '--cfg-path', '/var/folders/7_/1xx0hqcs3txd7kqt0ngfdjth0000gn/T/tmpx76emkve', '--no-shut-down-logging']\n[2023-01-17T13:42:19.101-0800] {standard_task_runner.py:84} INFO - Job 518: Subtask wait\n[2023-01-17T13:42:19.185-0800] {task_command.py:417} INFO - Running <TaskInstance: example_time_delta_sensor_async.wait manual__2023-01-17T21:41:38.150776+00:00 [running]> on host Daniels-MacBook-Pro-2.local\n[2023-01-17T13:42:19.274-0800] {taskinstance.py:1369} INFO - Marking task as SUCCESS. dag_id=example_time_delta_sensor_async, task_id=wait, execution_date=20230117T214138, start_date=20230117T214151, end_date=20230117T214219\n[2023-01-17T13:42:19.309-0800] {local_task_job.py:220} INFO - Task exited with return code 0\n[2023-01-17T13:42:19.339-0800] {taskinstance.py:2648} INFO - 1 downstream tasks scheduled from follow-on schedule check\n[2023-01-17T13:41:53.228-0800] {temporal.py:62} INFO - trigger starting\n[2023-01-17T13:41:53.228-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:41:54.228-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:41:55.229-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:41:56.229-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:41:57.229-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:41:58.230-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:41:59.231-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:42:00.231-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:42:01.232-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:42:02.233-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:42:03.233-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:42:04.233-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:42:05.233-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:42:06.234-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:42:07.235-0800] {temporal.py:71} INFO - sleeping 1 second...\n[2023-01-17T13:42:08.237-0800] {temporal.py:74} INFO - yielding event with payload DateTime(2023, 1, 17, 21, 42, 8, 150776, tzinfo=Timezone('UTC'))\n[2023-01-17T13:42:08.237-0800] {triggerer_job.py:540} INFO - Trigger <airflow.triggers.temporal.DateTimeTrigger moment=2023-01-17T21:42:08.150776+00:00> (ID 4) fired: TriggerEvent<DateTime(2023, 1, 17, 21, 42, 8, 150776, tzinfo=Timezone('UTC'))>\n"  
    print("\n".join(_interleave_logs(local_logs, remote_logs)))