Fix es/os task_hander convert_list_to_stream

Author: jason810496

providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py

@@ -46,10 +46,7 @@
 from airflow.providers.elasticsearch.log.es_response import ElasticSearchResponse, Hit
 from airflow.providers.elasticsearch.version_compat import AIRFLOW_V_3_0_PLUS
 from airflow.utils import timezone
-from airflow.utils.log.file_task_handler import (
-    FileTaskHandler,
-    convert_list_to_stream,
-)
+from airflow.utils.log.file_task_handler import FileTaskHandler
 from airflow.utils.log.logging_mixin import ExternalLoggingMixin, LoggingMixin
 from airflow.utils.module_loading import import_string
 from airflow.utils.session import create_session
@@ -355,9 +352,12 @@ def _read(
                     "Otherwise, the logs for this task instance may have been removed."
                 )
                 if AIRFLOW_V_3_0_PLUS:
-                    from airflow.utils.log.file_task_handler import StructuredLogMessage
+                    from airflow.utils.log.file_task_handler import (
+                        StructuredLogMessage,
+                        get_compatible_output_log_stream,
+                    )
 
-                    return convert_list_to_stream(
+                    return get_compatible_output_log_stream(
                         [
                             StructuredLogMessage(
                                 event=missing_log_message,
@@ -384,7 +384,10 @@ def concat_logs(hits: list[Hit]) -> str:
 
         if logs_by_host:
             if AIRFLOW_V_3_0_PLUS:
-                from airflow.utils.log.file_task_handler import StructuredLogMessage
+                from airflow.utils.log.file_task_handler import (
+                    StructuredLogMessage,
+                    get_compatible_output_log_stream,
+                )
 
                 header = [
                     StructuredLogMessage(
@@ -397,7 +400,7 @@ def concat_logs(hits: list[Hit]) -> str:
                 message = header + [
                     StructuredLogMessage(event=concat_logs(hits)) for hits in logs_by_host.values()
                 ]  # type: ignore[misc]
-                message = convert_list_to_stream(message)  # type: ignore[assignment]
+                message = get_compatible_output_log_stream(message)  # type: ignore[assignment]
             else:
                 message = [
                     (host, concat_logs(hits))  # type: ignore[misc]

providers/opensearch/src/airflow/providers/opensearch/log/os_task_handler.py

@@ -37,10 +37,7 @@
 from airflow.providers.opensearch.log.os_response import Hit, OpensearchResponse
 from airflow.providers.opensearch.version_compat import AIRFLOW_V_3_0_PLUS
 from airflow.utils import timezone
-from airflow.utils.log.file_task_handler import (
-    FileTaskHandler,
-    convert_list_to_stream,
-)
+from airflow.utils.log.file_task_handler import FileTaskHandler
 from airflow.utils.log.logging_mixin import ExternalLoggingMixin, LoggingMixin
 from airflow.utils.module_loading import import_string
 from airflow.utils.session import create_session
@@ -387,9 +384,12 @@ def _read(
                     "Otherwise, the logs for this task instance may have been removed."
                 )
                 if AIRFLOW_V_3_0_PLUS:
-                    from airflow.utils.log.file_task_handler import StructuredLogMessage
+                    from airflow.utils.log.file_task_handler import (
+                        StructuredLogMessage,
+                        get_compatible_output_log_stream,
+                    )
 
-                    return convert_list_to_stream(
+                    return get_compatible_output_log_stream(
                         [
                             StructuredLogMessage(
                                 event=missing_log_message,
@@ -416,7 +416,10 @@ def concat_logs(hits: list[Hit]):
 
         if logs_by_host:
             if AIRFLOW_V_3_0_PLUS:
-                from airflow.utils.log.file_task_handler import StructuredLogMessage
+                from airflow.utils.log.file_task_handler import (
+                    StructuredLogMessage,
+                    get_compatible_output_log_stream,
+                )
 
                 header = [
                     StructuredLogMessage(
@@ -429,7 +432,7 @@ def concat_logs(hits: list[Hit]):
                 message = header + [
                     StructuredLogMessage(event=concat_logs(hits)) for hits in logs_by_host.values()
                 ]
-                message = convert_list_to_stream(message)  # type: ignore[assignment]
+                message = get_compatible_output_log_stream(message)  # type: ignore[assignment]
             else:
                 message = [(host, concat_logs(hits)) for host, hits in logs_by_host.items()]  # type: ignore[misc]
         else: