DockerSwarmOperator: Support line breaks in service logs

airflow/providers/docker/operators/docker_swarm.py

@@ -207,7 +207,7 @@ def stream_new_logs(last_line_logged, since=0):
                 since=since,
                 timestamps=True,
             )
-            logs = b"".join(logs).decode().splitlines()
+            logs = list(map(lambda line: line.decode("utf-8"), logs))
             if last_line_logged in logs:
                 logs = logs[logs.index(last_line_logged) + 1 :]
             for line in logs:

tests/providers/docker/operators/test_docker_swarm.py

@@ -17,6 +17,7 @@
 # under the License.
 from __future__ import annotations
 
+import logging
 from unittest import mock
 
 import pytest
@@ -30,7 +31,7 @@
 
 class TestDockerSwarmOperator:
     @mock.patch("airflow.providers.docker.operators.docker_swarm.types")
-    def test_execute(self, types_mock, docker_api_client_patcher):
+    def test_execute(self, types_mock, docker_api_client_patcher, caplog):
         mock_obj = mock.Mock()
 
         def _client_tasks_side_effect():
@@ -40,7 +41,11 @@ def _client_tasks_side_effect():
                 yield [{"Status": {"State": "complete"}}]
 
         def _client_service_logs_effect():
-            yield b"2023-12-05T00:00:00.000000000Z Testing is awesome."
+            service_logs = [
+                b"2024-07-09T19:07:04.587918327Z lineone-one\rlineone-two",
+                b"2024-07-09T19:07:04.587918328Z linetwo",
+            ]
+            return (log for log in service_logs)
 
         client_mock = mock.Mock(spec=APIClient)
         client_mock.create_service.return_value = {"ID": "some_id"}
@@ -72,6 +77,7 @@ def _client_service_logs_effect():
             networks=["dummy_network"],
             placement=types.Placement(constraints=["node.labels.region==east"]),
         )
+        caplog.clear()
         operator.execute(None)
 
         types_mock.TaskTemplate.assert_called_once_with(
@@ -103,6 +109,14 @@ def _client_service_logs_effect():
             "some_id", follow=False, stdout=True, stderr=True, is_tty=True, since=0, timestamps=True
         )
 
+        with caplog.at_level(logging.INFO, logger=operator.log.name):
+            service_logs = [
+                "lineone-one\rlineone-two",
+                "linetwo",
+            ]
+            for log_line in service_logs:
+                assert log_line in caplog.messages
+
         csargs, cskwargs = client_mock.create_service.call_args_list[0]
         assert len(csargs) == 1, "create_service called with different number of arguments than expected"
         assert csargs == (mock_obj,)