Allow more empty loops before stopping log streaming (#52614)

jedcunningham · vatsrahul1001 · web-flow · commit 97bbf3ba031e · 2025-07-01T13:54:11.000+05:30
In #50715 we starting short-circuiting if we hit 5 iterations of no new log messages. This works well, except in the scenario where there are no log messages at all. ES log handler has it's own short-circuit for that scenario, but it triggers based on time and that works out to ~7 iterations. Let's let ES have the first crack at it so the user gets a better message. Co-authored-by: Rahul Vats <43964496+vatsrahul1001@users.noreply.github.com>
diff --git a/airflow-core/src/airflow/utils/log/log_reader.py b/airflow-core/src/airflow/utils/log/log_reader.py
@@ -45,7 +45,8 @@ class TaskLogReader:
 
     STREAM_LOOP_SLEEP_SECONDS = 1
     """Time to sleep between loops while waiting for more logs"""
-    STREAM_LOOP_STOP_AFTER_EMPTY_ITERATIONS = 5
+
+    STREAM_LOOP_STOP_AFTER_EMPTY_ITERATIONS = 10
     """Number of empty loop iterations before stopping the stream"""
 
     def read_log_chunks(
diff --git a/airflow-core/tests/unit/utils/log/test_log_reader.py b/airflow-core/tests/unit/utils/log/test_log_reader.py
@@ -222,11 +222,7 @@ def test_read_log_stream_should_read_each_try_in_turn(self, mock_read):
     def test_read_log_stream_no_end_of_log_marker(self, mock_read):
         mock_read.side_effect = [
             (["hello"], {"end_of_log": False}),
-            ([], {"end_of_log": False}),
-            ([], {"end_of_log": False}),
-            ([], {"end_of_log": False}),
-            ([], {"end_of_log": False}),
-            ([], {"end_of_log": False}),
+            *[([], {"end_of_log": False}) for _ in range(10)],
         ]
 
         self.ti.state = TaskInstanceState.SUCCESS
@@ -237,7 +233,7 @@ def test_read_log_stream_no_end_of_log_marker(self, mock_read):
             "hello\n",
             "(Log stream stopped - End of log marker not found; logs may be incomplete.)\n",
         ]
-        assert mock_read.call_count == 6
+        assert mock_read.call_count == 11
 
     def test_supports_external_link(self):
         task_log_reader = TaskLogReader()
