prompt: You will be provided with a partial code base and an issue statement explaining a problem to resolve.
  <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
  ```
  
  
  </issue>
  <code>
  [start of airflow/operators/bash_operator.py]
  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 of airflow/operators/bash_operator.py]
  </code>
  
  I need you to solve this issue by generating a single patch file that I can apply directly to this repository using git apply. Please respond with a single patch file in the following format.
  <patch>
  diff --git a/file.py b/file.py
  --- a/file.py
  +++ b/file.py
  @@ -1,27 +1,35 @@
   def euclidean(a, b):
  -    while b:
  -        a, b = b, a % b
  -    return a
  +    if b == 0:
  +        return a
  +    return euclidean(b, a % b)
   
   
   def bresenham(x0, y0, x1, y1):
       points = []
       dx = abs(x1 - x0)
       dy = abs(y1 - y0)
  -    sx = 1 if x0 < x1 else -1
  -    sy = 1 if y0 < y1 else -1
  -    err = dx - dy
  +    x, y = x0, y0
  +    sx = -1 if x0 > x1 else 1
  +    sy = -1 if y0 > y1 else 1
   
  -    while True:
  -        points.append((x0, y0))
  -        if x0 == x1 and y0 == y1:
  -            break
  -        e2 = 2 * err
  -        if e2 > -dy:
  -            err -= dy
  -            x0 += sx
  -        if e2 < dx:
  -            err += dx
  -            y0 += sy
  +    if dx > dy:
  +        err = dx / 2.0
  +        while x != x1:
  +            points.append((x, y))
  +            err -= dy
  +            if err < 0:
  +                y += sy
  +                err += dx
  +            x += sx
  +    else:
  +        err = dy / 2.0
  +        while y != y1:
  +            points.append((x, y))
  +            err -= dx
  +            if err < 0:
  +                x += sx
  +                err += dy
  +            y += sy
  +
  +    points.append((x, y))
       return points
  
  </patch>
golden_diff: 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 "
  
verification_info: {"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"}]}