We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
UnicodeDecodeError: 'utf-8' codec can't decode byte 0xff in position 369: invalid start byte
**Describe the bug**: 
Sending PUT request to FastAPI with binary file encoded, its returns a error.

**To Reproduce**

1. Configure FastAPI with elasticapm.contrib.starlette
2. Send a PUT request with binary file

**Environment (please complete the following information)**
- OS: Linux
- Python version: 3.8
- Framework and version: fastapi 0.61.2
- APM Server version: 8
- Agent version: 8

The same problema was resolved in the main code : [#344](https://github.com/elastic/apm-agent-python/issues/334)


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `elasticapm/contrib/starlette/utils.py`
Content:
```
1 #  BSD 3-Clause License
2 #
3 #  Copyright (c) 2019, Elasticsearch BV
4 #  All rights reserved.
5 #
6 #  Redistribution and use in source and binary forms, with or without
7 #  modification, are permitted provided that the following conditions are met:
8 #
9 #  * Redistributions of source code must retain the above copyright notice, this
10 #    list of conditions and the following disclaimer.
11 #
12 #  * Redistributions in binary form must reproduce the above copyright notice,
13 #    this list of conditions and the following disclaimer in the documentation
14 #    and/or other materials provided with the distribution.
15 #
16 #  * Neither the name of the copyright holder nor the names of its
17 #    contributors may be used to endorse or promote products derived from
18 #    this software without specific prior written permission.
19 #
20 #  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
21 #  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
22 #  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
23 #  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
24 #  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
25 #  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
26 #  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
27 #  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
28 #  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
29 #  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
30 import asyncio
31 
32 from starlette.datastructures import Headers
33 from starlette.requests import Request
34 from starlette.types import Message
35 
36 from elasticapm.conf import Config, constants
37 from elasticapm.utils import get_url_dict
38 
39 
40 async def get_data_from_request(request: Request, config: Config, event_type: str) -> dict:
41     """Loads data from incoming request for APM capturing.
42 
43     Args:
44         request (Request)
45         config (Config)
46         event_type (str)
47 
48     Returns:
49         dict
50     """
51     result = {
52         "method": request.method,
53         "socket": {"remote_address": _get_client_ip(request)},
54         "cookies": request.cookies,
55     }
56     if config.capture_headers:
57         result["headers"] = dict(request.headers)
58 
59     if request.method in constants.HTTP_WITH_BODY:
60         if config.capture_body not in ("all", event_type):
61             result["body"] = "[REDACTED]"
62         else:
63             body = None
64             try:
65                 body = await get_body(request)
66             except Exception:
67                 pass
68             if body is not None:
69                 result["body"] = body
70 
71     result["url"] = get_url_dict(str(request.url))
72 
73     return result
74 
75 
76 async def get_data_from_response(message: dict, config: Config, event_type: str) -> dict:
77     """Loads data from response for APM capturing.
78 
79     Args:
80         message (dict)
81         config (Config)
82         event_type (str)
83 
84     Returns:
85         dict
86     """
87     result = {}
88 
89     if "status" in message:
90         result["status_code"] = message["status"]
91 
92     if config.capture_headers and "headers" in message:
93         headers = Headers(raw=message["headers"])
94         result["headers"] = {key: ";".join(headers.getlist(key)) for key in headers.keys()}
95 
96     return result
97 
98 
99 async def set_body(request: Request, body: bytes):
100     """Overwrites body in Starlette.
101 
102     Args:
103         request (Request)
104         body (bytes)
105     """
106 
107     async def receive() -> Message:
108         await asyncio.sleep(0)
109         return {"type": "http.request", "body": body}
110 
111     request._receive = receive
112 
113 
114 async def get_body(request: Request) -> str:
115     """Gets body from the request.
116 
117     When we consume the body, we replace the streaming mechanism with
118     a mocked version -- this workaround came from
119     https://github.com/encode/starlette/issues/495#issuecomment-513138055
120 
121     Args:
122         request (Request)
123 
124     Returns:
125         str
126     """
127     body = await request.body()
128     await set_body(request, body)
129 
130     request._stream_consumed = False
131 
132     return body.decode("utf-8")
133 
134 
135 async def query_params_to_dict(query_params: str) -> dict:
136     """Transforms query params from URL to dictionary
137 
138     Args:
139         query_params (str)
140 
141     Returns:
142         dict
143 
144     Examples:
145         >>> print(query_params_to_dict(b"key=val&key2=val2"))
146         {"key": "val", "key2": "val2"}
147     """
148     query_params = query_params.split("&")
149     res = {}
150     for param in query_params:
151         key, val = param.split("=")
152         res[key] = val
153 
154     return res
155 
156 
157 def _get_client_ip(request: Request):
158     x_forwarded_for = request.headers.get("HTTP_X_FORWARDED_FOR")
159     if x_forwarded_for:
160         ip = x_forwarded_for.split(",")[0]
161     else:
162         ip = request.headers.get("REMOTE_ADDR")
163     return ip
164
```
--- 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/elasticapm/contrib/starlette/utils.py b/elasticapm/contrib/starlette/utils.py
--- a/elasticapm/contrib/starlette/utils.py
+++ b/elasticapm/contrib/starlette/utils.py
@@ -129,7 +129,7 @@
 
     request._stream_consumed = False
 
-    return body.decode("utf-8")
+    return body.decode("utf-8", errors="replace")
 
 
 async def query_params_to_dict(query_params: str) -> dict:

{"golden_diff": "diff --git a/elasticapm/contrib/starlette/utils.py b/elasticapm/contrib/starlette/utils.py\n--- a/elasticapm/contrib/starlette/utils.py\n+++ b/elasticapm/contrib/starlette/utils.py\n@@ -129,7 +129,7 @@\n \n     request._stream_consumed = False\n \n-    return body.decode(\"utf-8\")\n+    return body.decode(\"utf-8\", errors=\"replace\")\n \n \n async def query_params_to_dict(query_params: str) -> dict:\n", "issue": "UnicodeDecodeError: 'utf-8' codec can't decode byte 0xff in position 369: invalid start byte\n**Describe the bug**: \r\nSending PUT request to FastAPI with binary file encoded, its returns a error.\r\n\r\n**To Reproduce**\r\n\r\n1. Configure FastAPI with elasticapm.contrib.starlette\r\n2. Send a PUT request with binary file\r\n\r\n**Environment (please complete the following information)**\r\n- OS: Linux\r\n- Python version: 3.8\r\n- Framework and version: fastapi 0.61.2\r\n- APM Server version: 8\r\n- Agent version: 8\r\n\r\nThe same problema was resolved in the main code : [#344](https://github.com/elastic/apm-agent-python/issues/334)\r\n\n", "before_files": [{"content": "#  BSD 3-Clause License\n#\n#  Copyright (c) 2019, Elasticsearch BV\n#  All rights reserved.\n#\n#  Redistribution and use in source and binary forms, with or without\n#  modification, are permitted provided that the following conditions are met:\n#\n#  * Redistributions of source code must retain the above copyright notice, this\n#    list of conditions and the following disclaimer.\n#\n#  * Redistributions in binary form must reproduce the above copyright notice,\n#    this list of conditions and the following disclaimer in the documentation\n#    and/or other materials provided with the distribution.\n#\n#  * Neither the name of the copyright holder nor the names of its\n#    contributors may be used to endorse or promote products derived from\n#    this software without specific prior written permission.\n#\n#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS \"AS IS\"\n#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE\n#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE\n#  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE\n#  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL\n#  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR\n#  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER\n#  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,\n#  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE\n#  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\nimport asyncio\n\nfrom starlette.datastructures import Headers\nfrom starlette.requests import Request\nfrom starlette.types import Message\n\nfrom elasticapm.conf import Config, constants\nfrom elasticapm.utils import get_url_dict\n\n\nasync def get_data_from_request(request: Request, config: Config, event_type: str) -> dict:\n    \"\"\"Loads data from incoming request for APM capturing.\n\n    Args:\n        request (Request)\n        config (Config)\n        event_type (str)\n\n    Returns:\n        dict\n    \"\"\"\n    result = {\n        \"method\": request.method,\n        \"socket\": {\"remote_address\": _get_client_ip(request)},\n        \"cookies\": request.cookies,\n    }\n    if config.capture_headers:\n        result[\"headers\"] = dict(request.headers)\n\n    if request.method in constants.HTTP_WITH_BODY:\n        if config.capture_body not in (\"all\", event_type):\n            result[\"body\"] = \"[REDACTED]\"\n        else:\n            body = None\n            try:\n                body = await get_body(request)\n            except Exception:\n                pass\n            if body is not None:\n                result[\"body\"] = body\n\n    result[\"url\"] = get_url_dict(str(request.url))\n\n    return result\n\n\nasync def get_data_from_response(message: dict, config: Config, event_type: str) -> dict:\n    \"\"\"Loads data from response for APM capturing.\n\n    Args:\n        message (dict)\n        config (Config)\n        event_type (str)\n\n    Returns:\n        dict\n    \"\"\"\n    result = {}\n\n    if \"status\" in message:\n        result[\"status_code\"] = message[\"status\"]\n\n    if config.capture_headers and \"headers\" in message:\n        headers = Headers(raw=message[\"headers\"])\n        result[\"headers\"] = {key: \";\".join(headers.getlist(key)) for key in headers.keys()}\n\n    return result\n\n\nasync def set_body(request: Request, body: bytes):\n    \"\"\"Overwrites body in Starlette.\n\n    Args:\n        request (Request)\n        body (bytes)\n    \"\"\"\n\n    async def receive() -> Message:\n        await asyncio.sleep(0)\n        return {\"type\": \"http.request\", \"body\": body}\n\n    request._receive = receive\n\n\nasync def get_body(request: Request) -> str:\n    \"\"\"Gets body from the request.\n\n    When we consume the body, we replace the streaming mechanism with\n    a mocked version -- this workaround came from\n    https://github.com/encode/starlette/issues/495#issuecomment-513138055\n\n    Args:\n        request (Request)\n\n    Returns:\n        str\n    \"\"\"\n    body = await request.body()\n    await set_body(request, body)\n\n    request._stream_consumed = False\n\n    return body.decode(\"utf-8\")\n\n\nasync def query_params_to_dict(query_params: str) -> dict:\n    \"\"\"Transforms query params from URL to dictionary\n\n    Args:\n        query_params (str)\n\n    Returns:\n        dict\n\n    Examples:\n        >>> print(query_params_to_dict(b\"key=val&key2=val2\"))\n        {\"key\": \"val\", \"key2\": \"val2\"}\n    \"\"\"\n    query_params = query_params.split(\"&\")\n    res = {}\n    for param in query_params:\n        key, val = param.split(\"=\")\n        res[key] = val\n\n    return res\n\n\ndef _get_client_ip(request: Request):\n    x_forwarded_for = request.headers.get(\"HTTP_X_FORWARDED_FOR\")\n    if x_forwarded_for:\n        ip = x_forwarded_for.split(\",\")[0]\n    else:\n        ip = request.headers.get(\"REMOTE_ADDR\")\n    return ip\n", "path": "elasticapm/contrib/starlette/utils.py"}], "after_files": [{"content": "#  BSD 3-Clause License\n#\n#  Copyright (c) 2019, Elasticsearch BV\n#  All rights reserved.\n#\n#  Redistribution and use in source and binary forms, with or without\n#  modification, are permitted provided that the following conditions are met:\n#\n#  * Redistributions of source code must retain the above copyright notice, this\n#    list of conditions and the following disclaimer.\n#\n#  * Redistributions in binary form must reproduce the above copyright notice,\n#    this list of conditions and the following disclaimer in the documentation\n#    and/or other materials provided with the distribution.\n#\n#  * Neither the name of the copyright holder nor the names of its\n#    contributors may be used to endorse or promote products derived from\n#    this software without specific prior written permission.\n#\n#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS \"AS IS\"\n#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE\n#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE\n#  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE\n#  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL\n#  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR\n#  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER\n#  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,\n#  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE\n#  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\nimport asyncio\n\nfrom starlette.datastructures import Headers\nfrom starlette.requests import Request\nfrom starlette.types import Message\n\nfrom elasticapm.conf import Config, constants\nfrom elasticapm.utils import get_url_dict\n\n\nasync def get_data_from_request(request: Request, config: Config, event_type: str) -> dict:\n    \"\"\"Loads data from incoming request for APM capturing.\n\n    Args:\n        request (Request)\n        config (Config)\n        event_type (str)\n\n    Returns:\n        dict\n    \"\"\"\n    result = {\n        \"method\": request.method,\n        \"socket\": {\"remote_address\": _get_client_ip(request)},\n        \"cookies\": request.cookies,\n    }\n    if config.capture_headers:\n        result[\"headers\"] = dict(request.headers)\n\n    if request.method in constants.HTTP_WITH_BODY:\n        if config.capture_body not in (\"all\", event_type):\n            result[\"body\"] = \"[REDACTED]\"\n        else:\n            body = None\n            try:\n                body = await get_body(request)\n            except Exception:\n                pass\n            if body is not None:\n                result[\"body\"] = body\n\n    result[\"url\"] = get_url_dict(str(request.url))\n\n    return result\n\n\nasync def get_data_from_response(message: dict, config: Config, event_type: str) -> dict:\n    \"\"\"Loads data from response for APM capturing.\n\n    Args:\n        message (dict)\n        config (Config)\n        event_type (str)\n\n    Returns:\n        dict\n    \"\"\"\n    result = {}\n\n    if \"status\" in message:\n        result[\"status_code\"] = message[\"status\"]\n\n    if config.capture_headers and \"headers\" in message:\n        headers = Headers(raw=message[\"headers\"])\n        result[\"headers\"] = {key: \";\".join(headers.getlist(key)) for key in headers.keys()}\n\n    return result\n\n\nasync def set_body(request: Request, body: bytes):\n    \"\"\"Overwrites body in Starlette.\n\n    Args:\n        request (Request)\n        body (bytes)\n    \"\"\"\n\n    async def receive() -> Message:\n        await asyncio.sleep(0)\n        return {\"type\": \"http.request\", \"body\": body}\n\n    request._receive = receive\n\n\nasync def get_body(request: Request) -> str:\n    \"\"\"Gets body from the request.\n\n    When we consume the body, we replace the streaming mechanism with\n    a mocked version -- this workaround came from\n    https://github.com/encode/starlette/issues/495#issuecomment-513138055\n\n    Args:\n        request (Request)\n\n    Returns:\n        str\n    \"\"\"\n    body = await request.body()\n    await set_body(request, body)\n\n    request._stream_consumed = False\n\n    return body.decode(\"utf-8\", errors=\"replace\")\n\n\nasync def query_params_to_dict(query_params: str) -> dict:\n    \"\"\"Transforms query params from URL to dictionary\n\n    Args:\n        query_params (str)\n\n    Returns:\n        dict\n\n    Examples:\n        >>> print(query_params_to_dict(b\"key=val&key2=val2\"))\n        {\"key\": \"val\", \"key2\": \"val2\"}\n    \"\"\"\n    query_params = query_params.split(\"&\")\n    res = {}\n    for param in query_params:\n        key, val = param.split(\"=\")\n        res[key] = val\n\n    return res\n\n\ndef _get_client_ip(request: Request):\n    x_forwarded_for = request.headers.get(\"HTTP_X_FORWARDED_FOR\")\n    if x_forwarded_for:\n        ip = x_forwarded_for.split(\",\")[0]\n    else:\n        ip = request.headers.get(\"REMOTE_ADDR\")\n    return ip\n", "path": "elasticapm/contrib/starlette/utils.py"}]}