We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
UnicodeDecodeError in bash_operator.py
Hi,

I see a lot of these errors when running `airflow backfill` : 

```
Traceback (most recent call last):
  File "/usr/lib/python2.7/logging/__init__.py", line 851, in emit
    msg = self.format(record)
  File "/usr/lib/python2.7/logging/__init__.py", line 724, in format
    return fmt.format(record)
  File "/usr/lib/python2.7/logging/__init__.py", line 467, in format
    s = self._fmt % record.__dict__
UnicodeDecodeError: 'ascii' codec can't decode byte 0xc3 in position 13: ordinal not in range(128)
Logged from file bash_operator.py, line 72
```


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `airflow/operators/bash_operator.py`
Content:
```
1 
2 from builtins import bytes
3 import logging
4 import sys
5 from subprocess import Popen, STDOUT, PIPE
6 from tempfile import gettempdir, NamedTemporaryFile
7 
8 from airflow.utils import AirflowException
9 from airflow.models import BaseOperator
10 from airflow.utils import apply_defaults, TemporaryDirectory
11 
12 
13 class BashOperator(BaseOperator):
14     """
15     Execute a Bash script, command or set of commands.
16 
17     :param bash_command: The command, set of commands or reference to a
18         bash script (must be '.sh') to be executed.
19     :type bash_command: string
20     :param env: If env is not None, it must be a mapping that defines the
21         environment variables for the new process; these are used instead
22         of inheriting the current process environment, which is the default
23         behavior.
24     :type env: dict
25     """
26     template_fields = ('bash_command', 'env')
27     template_ext = ('.sh', '.bash',)
28     ui_color = '#f0ede4'
29 
30     @apply_defaults
31     def __init__(
32             self,
33             bash_command,
34             xcom_push=False,
35             env=None,
36             *args, **kwargs):
37         """
38         If xcom_push is True, the last line written to stdout will also
39         be pushed to an XCom when the bash command completes.
40         """
41         super(BashOperator, self).__init__(*args, **kwargs)
42         self.bash_command = bash_command
43         self.env = env
44         self.xcom_push_flag = xcom_push
45 
46     def execute(self, context):
47         """
48         Execute the bash command in a temporary directory
49         which will be cleaned afterwards
50         """
51         bash_command = self.bash_command
52         logging.info("tmp dir root location: \n" + gettempdir())
53         with TemporaryDirectory(prefix='airflowtmp') as tmp_dir:
54             with NamedTemporaryFile(dir=tmp_dir, prefix=self.task_id) as f:
55 
56                 f.write(bytes(bash_command, 'utf_8'))
57                 f.flush()
58                 fname = f.name
59                 script_location = tmp_dir + "/" + fname
60                 logging.info("Temporary script "
61                              "location :{0}".format(script_location))
62                 logging.info("Running command: " + bash_command)
63                 sp = Popen(
64                     ['bash', fname],
65                     stdout=PIPE, stderr=STDOUT,
66                     cwd=tmp_dir, env=self.env)
67 
68                 self.sp = sp
69 
70                 logging.info("Output:")
71                 line = ''
72                 for line in iter(sp.stdout.readline, b''):
73                     line = line.decode().strip()
74                     logging.info(line)
75                 sp.wait()
76                 logging.info("Command exited with "
77                              "return code {0}".format(sp.returncode))
78 
79                 if sp.returncode:
80                     raise AirflowException("Bash command failed")
81 
82         if self.xcom_push_flag:
83             return line
84 
85     def on_kill(self):
86         logging.info('Sending SIGTERM signal to bash subprocess')
87         self.sp.terminate()
88
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/airflow/operators/bash_operator.py b/airflow/operators/bash_operator.py
--- a/airflow/operators/bash_operator.py
+++ b/airflow/operators/bash_operator.py
@@ -1,7 +1,6 @@
 
 from builtins import bytes
 import logging
-import sys
 from subprocess import Popen, STDOUT, PIPE
 from tempfile import gettempdir, NamedTemporaryFile
 
@@ -22,6 +21,7 @@
         of inheriting the current process environment, which is the default
         behavior.
     :type env: dict
+    :type output_encoding: output encoding of bash command
     """
     template_fields = ('bash_command', 'env')
     template_ext = ('.sh', '.bash',)
@@ -33,6 +33,7 @@
             bash_command,
             xcom_push=False,
             env=None,
+            output_encoding='utf-8',
             *args, **kwargs):
         """
         If xcom_push is True, the last line written to stdout will also
@@ -42,6 +43,7 @@
         self.bash_command = bash_command
         self.env = env
         self.xcom_push_flag = xcom_push
+        self.output_encoding = output_encoding
 
     def execute(self, context):
         """
@@ -70,7 +72,7 @@
                 logging.info("Output:")
                 line = ''
                 for line in iter(sp.stdout.readline, b''):
-                    line = line.decode().strip()
+                    line = line.decode(self.output_encoding).strip()
                     logging.info(line)
                 sp.wait()
                 logging.info("Command exited with "

{"golden_diff": "diff --git a/airflow/operators/bash_operator.py b/airflow/operators/bash_operator.py\n--- a/airflow/operators/bash_operator.py\n+++ b/airflow/operators/bash_operator.py\n@@ -1,7 +1,6 @@\n \n from builtins import bytes\n import logging\n-import sys\n from subprocess import Popen, STDOUT, PIPE\n from tempfile import gettempdir, NamedTemporaryFile\n \n@@ -22,6 +21,7 @@\n         of inheriting the current process environment, which is the default\n         behavior.\n     :type env: dict\n+    :type output_encoding: output encoding of bash command\n     \"\"\"\n     template_fields = ('bash_command', 'env')\n     template_ext = ('.sh', '.bash',)\n@@ -33,6 +33,7 @@\n             bash_command,\n             xcom_push=False,\n             env=None,\n+            output_encoding='utf-8',\n             *args, **kwargs):\n         \"\"\"\n         If xcom_push is True, the last line written to stdout will also\n@@ -42,6 +43,7 @@\n         self.bash_command = bash_command\n         self.env = env\n         self.xcom_push_flag = xcom_push\n+        self.output_encoding = output_encoding\n \n     def execute(self, context):\n         \"\"\"\n@@ -70,7 +72,7 @@\n                 logging.info(\"Output:\")\n                 line = ''\n                 for line in iter(sp.stdout.readline, b''):\n-                    line = line.decode().strip()\n+                    line = line.decode(self.output_encoding).strip()\n                     logging.info(line)\n                 sp.wait()\n                 logging.info(\"Command exited with \"\n", "issue": "UnicodeDecodeError in bash_operator.py\nHi,\n\nI see a lot of these errors when running `airflow backfill` : \n\n```\nTraceback (most recent call last):\n  File \"/usr/lib/python2.7/logging/__init__.py\", line 851, in emit\n    msg = self.format(record)\n  File \"/usr/lib/python2.7/logging/__init__.py\", line 724, in format\n    return fmt.format(record)\n  File \"/usr/lib/python2.7/logging/__init__.py\", line 467, in format\n    s = self._fmt % record.__dict__\nUnicodeDecodeError: 'ascii' codec can't decode byte 0xc3 in position 13: ordinal not in range(128)\nLogged from file bash_operator.py, line 72\n```\n\n", "before_files": [{"content": "\nfrom builtins import bytes\nimport logging\nimport sys\nfrom subprocess import Popen, STDOUT, PIPE\nfrom tempfile import gettempdir, NamedTemporaryFile\n\nfrom airflow.utils import AirflowException\nfrom airflow.models import BaseOperator\nfrom airflow.utils import apply_defaults, TemporaryDirectory\n\n\nclass BashOperator(BaseOperator):\n    \"\"\"\n    Execute a Bash script, command or set of commands.\n\n    :param bash_command: The command, set of commands or reference to a\n        bash script (must be '.sh') to be executed.\n    :type bash_command: string\n    :param env: If env is not None, it must be a mapping that defines the\n        environment variables for the new process; these are used instead\n        of inheriting the current process environment, which is the default\n        behavior.\n    :type env: dict\n    \"\"\"\n    template_fields = ('bash_command', 'env')\n    template_ext = ('.sh', '.bash',)\n    ui_color = '#f0ede4'\n\n    @apply_defaults\n    def __init__(\n            self,\n            bash_command,\n            xcom_push=False,\n            env=None,\n            *args, **kwargs):\n        \"\"\"\n        If xcom_push is True, the last line written to stdout will also\n        be pushed to an XCom when the bash command completes.\n        \"\"\"\n        super(BashOperator, self).__init__(*args, **kwargs)\n        self.bash_command = bash_command\n        self.env = env\n        self.xcom_push_flag = xcom_push\n\n    def execute(self, context):\n        \"\"\"\n        Execute the bash command in a temporary directory\n        which will be cleaned afterwards\n        \"\"\"\n        bash_command = self.bash_command\n        logging.info(\"tmp dir root location: \\n\" + gettempdir())\n        with TemporaryDirectory(prefix='airflowtmp') as tmp_dir:\n            with NamedTemporaryFile(dir=tmp_dir, prefix=self.task_id) as f:\n\n                f.write(bytes(bash_command, 'utf_8'))\n                f.flush()\n                fname = f.name\n                script_location = tmp_dir + \"/\" + fname\n                logging.info(\"Temporary script \"\n                             \"location :{0}\".format(script_location))\n                logging.info(\"Running command: \" + bash_command)\n                sp = Popen(\n                    ['bash', fname],\n                    stdout=PIPE, stderr=STDOUT,\n                    cwd=tmp_dir, env=self.env)\n\n                self.sp = sp\n\n                logging.info(\"Output:\")\n                line = ''\n                for line in iter(sp.stdout.readline, b''):\n                    line = line.decode().strip()\n                    logging.info(line)\n                sp.wait()\n                logging.info(\"Command exited with \"\n                             \"return code {0}\".format(sp.returncode))\n\n                if sp.returncode:\n                    raise AirflowException(\"Bash command failed\")\n\n        if self.xcom_push_flag:\n            return line\n\n    def on_kill(self):\n        logging.info('Sending SIGTERM signal to bash subprocess')\n        self.sp.terminate()\n", "path": "airflow/operators/bash_operator.py"}], "after_files": [{"content": "\nfrom builtins import bytes\nimport logging\nfrom subprocess import Popen, STDOUT, PIPE\nfrom tempfile import gettempdir, NamedTemporaryFile\n\nfrom airflow.utils import AirflowException\nfrom airflow.models import BaseOperator\nfrom airflow.utils import apply_defaults, TemporaryDirectory\n\n\nclass BashOperator(BaseOperator):\n    \"\"\"\n    Execute a Bash script, command or set of commands.\n\n    :param bash_command: The command, set of commands or reference to a\n        bash script (must be '.sh') to be executed.\n    :type bash_command: string\n    :param env: If env is not None, it must be a mapping that defines the\n        environment variables for the new process; these are used instead\n        of inheriting the current process environment, which is the default\n        behavior.\n    :type env: dict\n    :type output_encoding: output encoding of bash command\n    \"\"\"\n    template_fields = ('bash_command', 'env')\n    template_ext = ('.sh', '.bash',)\n    ui_color = '#f0ede4'\n\n    @apply_defaults\n    def __init__(\n            self,\n            bash_command,\n            xcom_push=False,\n            env=None,\n            output_encoding='utf-8',\n            *args, **kwargs):\n        \"\"\"\n        If xcom_push is True, the last line written to stdout will also\n        be pushed to an XCom when the bash command completes.\n        \"\"\"\n        super(BashOperator, self).__init__(*args, **kwargs)\n        self.bash_command = bash_command\n        self.env = env\n        self.xcom_push_flag = xcom_push\n        self.output_encoding = output_encoding\n\n    def execute(self, context):\n        \"\"\"\n        Execute the bash command in a temporary directory\n        which will be cleaned afterwards\n        \"\"\"\n        bash_command = self.bash_command\n        logging.info(\"tmp dir root location: \\n\" + gettempdir())\n        with TemporaryDirectory(prefix='airflowtmp') as tmp_dir:\n            with NamedTemporaryFile(dir=tmp_dir, prefix=self.task_id) as f:\n\n                f.write(bytes(bash_command, 'utf_8'))\n                f.flush()\n                fname = f.name\n                script_location = tmp_dir + \"/\" + fname\n                logging.info(\"Temporary script \"\n                             \"location :{0}\".format(script_location))\n                logging.info(\"Running command: \" + bash_command)\n                sp = Popen(\n                    ['bash', fname],\n                    stdout=PIPE, stderr=STDOUT,\n                    cwd=tmp_dir, env=self.env)\n\n                self.sp = sp\n\n                logging.info(\"Output:\")\n                line = ''\n                for line in iter(sp.stdout.readline, b''):\n                    line = line.decode(self.output_encoding).strip()\n                    logging.info(line)\n                sp.wait()\n                logging.info(\"Command exited with \"\n                             \"return code {0}\".format(sp.returncode))\n\n                if sp.returncode:\n                    raise AirflowException(\"Bash command failed\")\n\n        if self.xcom_push_flag:\n            return line\n\n    def on_kill(self):\n        logging.info('Sending SIGTERM signal to bash subprocess')\n        self.sp.terminate()\n", "path": "airflow/operators/bash_operator.py"}]}