Fix: Improve ingestion logging for streaming logs (#27213)

* Fix: Improve ingestion logging for streaming logs

* Apply suggestion from @Copilot

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* test: verify print_status runs before stop in execute() teardown

Covers the ordering fix and finally-guarantee: print_status must run
before stop (so records are flushed while resources are alive), and
stop must still run when print_status raises.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: 3 people

ingestion/src/metadata/workflow/base.py

@@ -281,8 +281,10 @@ def execute(self) -> None:
         finally:
             ingestion_status = self.build_ingestion_status()
             self.set_ingestion_pipeline_status(pipeline_state, ingestion_status)
-            self.stop()
-            self.print_status()
+            try:
+                self.print_status()
+            finally:
+                self.stop()
 
     @property
     def run_id(self) -> str:

ingestion/tests/unit/workflow/test_base_workflow.py

@@ -13,6 +13,7 @@
 """
 from typing import Iterable, Tuple
 from unittest import TestCase
+from unittest.mock import MagicMock, patch
 
 import pytest
 
@@ -200,3 +201,45 @@ def test_workflow_config_supports_ingestion_runner_name(self):
         )
 
         self.assertEqual(workflow_config.ingestionRunnerName, "test-runner")
+
+
+class TestWorkflowExecuteTeardown:
+    """
+    Validates the execute() teardown contract: status must be printed before
+    stop() tears down resources (so final records are flushed while the
+    metadata client and steps are alive), and stop() must still run when
+    print_status() raises so we never leak the timer thread or OM client.
+    """
+
+    def test_print_status_runs_before_stop(self):
+        workflow = SimpleWorkflow(config=config)
+        manager = MagicMock()
+
+        with patch.object(
+            workflow, "print_status", wraps=workflow.print_status
+        ) as mock_print_status, patch.object(
+            workflow, "stop", wraps=workflow.stop
+        ) as mock_stop:
+            manager.attach_mock(mock_print_status, "print_status")
+            manager.attach_mock(mock_stop, "stop")
+
+            workflow.execute()
+
+        ordered_names = [mock_call[0] for mock_call in manager.mock_calls]
+        assert ordered_names == ["print_status", "stop"]
+
+    def test_stop_still_runs_when_print_status_raises(self):
+        workflow = SimpleWorkflow(config=config)
+
+        with patch.object(
+            workflow,
+            "print_status",
+            side_effect=RuntimeError("boom"),
+        ) as mock_print_status, patch.object(
+            workflow, "stop", wraps=workflow.stop
+        ) as mock_stop:
+            with pytest.raises(RuntimeError, match="boom"):
+                workflow.execute()
+
+            mock_print_status.assert_called_once()
+            mock_stop.assert_called_once()