[v3-1-test] Emit log stream stopped warning as ndjson (#56474) (#56480)

Changes the format of the log stream stopped warning emitted by the TaskLogReader to ndjson when it encounters no end of log marker in a stream. Mixing ndjson and non-ndjson means the UI will not show any logs at all.
(cherry picked from commit 8782650)

Co-authored-by: Ian Buss <[email protected]>

Author: github-actions[bot]

airflow-core/src/airflow/utils/log/log_reader.py

@@ -147,7 +147,8 @@ def read_log_stream(
                     empty_iterations += 1
                     if empty_iterations >= self.STREAM_LOOP_STOP_AFTER_EMPTY_ITERATIONS:
                         # we have not received any logs for a while, so we stop the stream
-                        yield "(Log stream stopped - End of log marker not found; logs may be incomplete.)\n"
+                        # this is emitted as json to avoid breaking the ndjson stream format
+                        yield '{"event": "Log stream stopped - End of log marker not found; logs may be incomplete."}\n'
                         return
             else:
                 # https://mypy.readthedocs.io/en/stable/typed_dict.html#supported-operations

airflow-core/tests/unit/utils/log/test_log_reader.py

@@ -268,7 +268,7 @@ def test_read_log_stream_no_end_of_log_marker(self, mock_read):
         log_stream = task_log_reader.read_log_stream(ti=self.ti, try_number=1, metadata={})
         assert list(log_stream) == [
             '{"timestamp":null,"event":"hello"}\n',
-            "(Log stream stopped - End of log marker not found; logs may be incomplete.)\n",
+            '{"event": "Log stream stopped - End of log marker not found; logs may be incomplete."}\n',
         ]
         assert mock_read.call_count == 11