Fix string encoding in DockerOperator when using xcom / json

airflow/providers/docker/CHANGELOG.rst

@@ -26,7 +26,8 @@ Features
 ~~~~~~~~
 
 * ``Adds option to disable mounting temporary folder in DockerOperator (#16932)``
-
+* if ``xcom_all`` is set to ``False``, only the last line of the log (separated by ``\n``) will be
+  included in the XCom value
 
 Bug Fixes
 ~~~~~~~~~
@@ -38,6 +39,8 @@ That was an unintended side effect of #15843 that has been fixed in #16932. Ther
 which will make Docker Operator works with warning and you will be able to remove the warning by
 using the new parameter to disable mounting the folder.
 
+* the return value of XCom is of type ``str`` (as opposed to ``bytes``)
+
 .. Below changes are excluded from the changelog. Move them to
    appropriate section above if needed. Do not delete the lines(!):
    * ``Removes pylint from our toolchain (#16682)``

airflow/providers/docker/operators/docker.py

@@ -299,22 +299,19 @@ def _run_image_with_mounts(self, target_mounts, add_tmp_variable: bool) -> Optio
             line = ''
             res_lines = []
             for line in lines:
-                line = line.strip()
                 if hasattr(line, 'decode'):
                     # Note that lines returned can also be byte sequences so we have to handle decode here
                     line = line.decode('utf-8')
+                line = line.strip()
                 res_lines.append(line)
                 self.log.info(line)
 
             result = self.cli.wait(self.container['Id'])
             if result['StatusCode'] != 0:
                 res_lines = "\n".join(res_lines)
                 raise AirflowException('docker container failed: ' + repr(result) + f"lines {res_lines}")
-
-            ret = None
             if self.do_xcom_push:
-                ret = self.cli.logs(container=self.container['Id']) if self.xcom_all else line.encode('utf-8')
-            return ret
+                return res_lines if self.xcom_all else line
         finally:
             if self.auto_remove:
                 self.cli.remove_container(self.container['Id'])

tests/providers/docker/operators/test_docker.py

@@ -47,8 +47,7 @@ def setUp(self):
         self.client_mock = mock.Mock(spec=APIClient)
         self.client_mock.create_container.return_value = {'Id': 'some_id'}
         self.client_mock.images.return_value = []
-        self.client_mock.attach.return_value = ['container log']
-        self.client_mock.logs.return_value = ['container log']
+        self.client_mock.attach.return_value = ['container log 1', 'container log 2']
         self.client_mock.pull.return_value = {"status": "pull log"}
         self.client_mock.wait.return_value = {"StatusCode": 0}
         self.client_mock.create_host_config.return_value = mock.Mock()
@@ -421,13 +420,48 @@ def test_execute_xcom_behavior(self):
             'tty': True,
         }
 
-        xcom_push_operator = DockerOperator(**kwargs, do_xcom_push=True)
+        xcom_push_operator = DockerOperator(**kwargs, do_xcom_push=True, xcom_all=False)
+        xcom_all_operator = DockerOperator(**kwargs, do_xcom_push=True, xcom_all=True)
         no_xcom_push_operator = DockerOperator(**kwargs, do_xcom_push=False)
 
         xcom_push_result = xcom_push_operator.execute(None)
+        xcom_all_result = xcom_all_operator.execute(None)
         no_xcom_push_result = no_xcom_push_operator.execute(None)
 
-        assert xcom_push_result == b'container log'
+        assert xcom_push_result == 'container log 2'
+        assert xcom_all_result == ['container log 1', 'container log 2']
+        assert no_xcom_push_result is None
+
+    def test_execute_xcom_behavior_bytes(self):
+        self.client_mock.pull.return_value = [b'{"status":"pull log"}']
+        self.client_mock.attach.return_value = [b'container log 1 ', b'container log 2']
+        kwargs = {
+            'api_version': '1.19',
+            'command': 'env',
+            'environment': {'UNIT': 'TEST'},
+            'private_environment': {'PRIVATE': 'MESSAGE'},
+            'image': 'ubuntu:latest',
+            'network_mode': 'bridge',
+            'owner': 'unittest',
+            'task_id': 'unittest',
+            'mounts': [Mount(source='/host/path', target='/container/path', type='bind')],
+            'working_dir': '/container/path',
+            'shm_size': 1000,
+            'host_tmp_dir': '/host/airflow',
+            'container_name': 'test_container',
+            'tty': True,
+        }
+
+        xcom_push_operator = DockerOperator(**kwargs, do_xcom_push=True, xcom_all=False)
+        xcom_all_operator = DockerOperator(**kwargs, do_xcom_push=True, xcom_all=True)
+        no_xcom_push_operator = DockerOperator(**kwargs, do_xcom_push=False)
+
+        xcom_push_result = xcom_push_operator.execute(None)
+        xcom_all_result = xcom_all_operator.execute(None)
+        no_xcom_push_result = no_xcom_push_operator.execute(None)
+
+        assert xcom_push_result == 'container log 2'
+        assert xcom_all_result == ['container log 1', 'container log 2']
         assert no_xcom_push_result is None
 
     def test_extra_hosts(self):