We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
'AsyncTransport' has no attribute '_start_event_processor'
**Describe the bug**: ...

After upgrading to `elastic-apm==5.4.1` I now get an error when Celery starts:

```
<function _register_worker_signals.<locals>.worker_startup at 0x7feae4beb620> raised: AttributeError("'AsyncTransport' object has no attribute '_start_event_processor'",)
Traceback (most recent call last):
  File "/venv/lib/python3.6/site-packages/celery/utils/dispatch/signal.py", line 288, in send
    response = receiver(signal=self, sender=sender, **named)
  File "/venv/1.37.1/lib/python3.6/site-packages/elasticapm/contrib/celery/__init__.py", line 80, in worker_startup
    client._transport._start_event_processor()
AttributeError: 'AsyncTransport' object has no attribute '_start_event_processor' 
```

**Environment (please complete the following information)**
- OS: Linux-3.10.0-1062.9.1.el7.x86_64-x86_64-with-centos-7.7.1908-Core 2020-02-18 16:24:31
- Python version: Python 3.6.8
- Framework and version: 
  celery 4.4.0
  Django 3.0.3
- APM Server version: ?
- Agent version: 5.4.1

I see the same error mentioned in issue #704, but I don't seem to have an issue with restarting Celery workers.

--- 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/celery/__init__.py`
Content:
```
1 #  BSD 3-Clause License
2 #
3 #  Copyright (c) 2012, the Sentry Team, see AUTHORS for more details
4 #  Copyright (c) 2019, Elasticsearch BV
5 #  All rights reserved.
6 #
7 #  Redistribution and use in source and binary forms, with or without
8 #  modification, are permitted provided that the following conditions are met:
9 #
10 #  * Redistributions of source code must retain the above copyright notice, this
11 #    list of conditions and the following disclaimer.
12 #
13 #  * Redistributions in binary form must reproduce the above copyright notice,
14 #    this list of conditions and the following disclaimer in the documentation
15 #    and/or other materials provided with the distribution.
16 #
17 #  * Neither the name of the copyright holder nor the names of its
18 #    contributors may be used to endorse or promote products derived from
19 #    this software without specific prior written permission.
20 #
21 #  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
22 #  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
23 #  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
24 #  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
25 #  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
26 #  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
27 #  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
28 #  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
29 #  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
30 
31 
32 from celery import signals
33 
34 from elasticapm.utils import get_name_from_func
35 
36 
37 class CeleryFilter(object):
38     def filter(self, record):
39         if record.funcName in ("_log_error",):
40             return 0
41         else:
42             return 1
43 
44 
45 def register_exception_tracking(client):
46     dispatch_uid = "elasticapm-exc-tracking"
47 
48     def process_failure_signal(sender, task_id, exception, args, kwargs, traceback, einfo, **kw):
49         client.capture_exception(
50             extra={"task_id": task_id, "task": sender, "args": args, "kwargs": kwargs}, handled=False
51         )
52 
53     signals.task_failure.disconnect(process_failure_signal, dispatch_uid=dispatch_uid)
54     signals.task_failure.connect(process_failure_signal, weak=False, dispatch_uid=dispatch_uid)
55     _register_worker_signals(client)
56 
57 
58 def register_instrumentation(client):
59     def begin_transaction(*args, **kwargs):
60         client.begin_transaction("celery")
61 
62     def end_transaction(task_id, task, *args, **kwargs):
63         name = get_name_from_func(task)
64         client.end_transaction(name, kwargs.get("state", "None"))
65 
66     dispatch_uid = "elasticapm-tracing-%s"
67 
68     # unregister any existing clients
69     signals.task_prerun.disconnect(begin_transaction, dispatch_uid=dispatch_uid % "prerun")
70     signals.task_postrun.disconnect(end_transaction, dispatch_uid=dispatch_uid % "postrun")
71 
72     # register for this client
73     signals.task_prerun.connect(begin_transaction, dispatch_uid=dispatch_uid % "prerun", weak=False)
74     signals.task_postrun.connect(end_transaction, weak=False, dispatch_uid=dispatch_uid % "postrun")
75     _register_worker_signals(client)
76 
77 
78 def _register_worker_signals(client):
79     def worker_startup(*args, **kwargs):
80         client._transport._start_event_processor()
81 
82     def worker_shutdown(*args, **kwargs):
83         client.close()
84 
85     def connect_worker_process_init(*args, **kwargs):
86         signals.worker_process_init.connect(worker_startup, dispatch_uid="elasticapm-start-worker", weak=False)
87         signals.worker_process_shutdown.connect(worker_shutdown, dispatch_uid="elasticapm-shutdown-worker", weak=False)
88 
89     signals.worker_init.connect(
90         connect_worker_process_init, dispatch_uid="elasticapm-connect-start-threads", weak=False
91     )
92
```
--- 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/celery/__init__.py b/elasticapm/contrib/celery/__init__.py
--- a/elasticapm/contrib/celery/__init__.py
+++ b/elasticapm/contrib/celery/__init__.py
@@ -76,14 +76,10 @@
 
 
 def _register_worker_signals(client):
-    def worker_startup(*args, **kwargs):
-        client._transport._start_event_processor()
-
     def worker_shutdown(*args, **kwargs):
         client.close()
 
     def connect_worker_process_init(*args, **kwargs):
-        signals.worker_process_init.connect(worker_startup, dispatch_uid="elasticapm-start-worker", weak=False)
         signals.worker_process_shutdown.connect(worker_shutdown, dispatch_uid="elasticapm-shutdown-worker", weak=False)
 
     signals.worker_init.connect(

{"golden_diff": "diff --git a/elasticapm/contrib/celery/__init__.py b/elasticapm/contrib/celery/__init__.py\n--- a/elasticapm/contrib/celery/__init__.py\n+++ b/elasticapm/contrib/celery/__init__.py\n@@ -76,14 +76,10 @@\n \n \n def _register_worker_signals(client):\n-    def worker_startup(*args, **kwargs):\n-        client._transport._start_event_processor()\n-\n     def worker_shutdown(*args, **kwargs):\n         client.close()\n \n     def connect_worker_process_init(*args, **kwargs):\n-        signals.worker_process_init.connect(worker_startup, dispatch_uid=\"elasticapm-start-worker\", weak=False)\n         signals.worker_process_shutdown.connect(worker_shutdown, dispatch_uid=\"elasticapm-shutdown-worker\", weak=False)\n \n     signals.worker_init.connect(\n", "issue": "'AsyncTransport' has no attribute '_start_event_processor'\n**Describe the bug**: ...\r\n\r\nAfter upgrading to `elastic-apm==5.4.1` I now get an error when Celery starts:\r\n\r\n```\r\n<function _register_worker_signals.<locals>.worker_startup at 0x7feae4beb620> raised: AttributeError(\"'AsyncTransport' object has no attribute '_start_event_processor'\",)\r\nTraceback (most recent call last):\r\n  File \"/venv/lib/python3.6/site-packages/celery/utils/dispatch/signal.py\", line 288, in send\r\n    response = receiver(signal=self, sender=sender, **named)\r\n  File \"/venv/1.37.1/lib/python3.6/site-packages/elasticapm/contrib/celery/__init__.py\", line 80, in worker_startup\r\n    client._transport._start_event_processor()\r\nAttributeError: 'AsyncTransport' object has no attribute '_start_event_processor' \r\n```\r\n\r\n**Environment (please complete the following information)**\r\n- OS: Linux-3.10.0-1062.9.1.el7.x86_64-x86_64-with-centos-7.7.1908-Core 2020-02-18 16:24:31\r\n- Python version: Python 3.6.8\r\n- Framework and version: \r\n  celery 4.4.0\r\n  Django 3.0.3\r\n- APM Server version: ?\r\n- Agent version: 5.4.1\r\n\r\nI see the same error mentioned in issue #704, but I don't seem to have an issue with restarting Celery workers.\n", "before_files": [{"content": "#  BSD 3-Clause License\n#\n#  Copyright (c) 2012, the Sentry Team, see AUTHORS for more details\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\n\nfrom celery import signals\n\nfrom elasticapm.utils import get_name_from_func\n\n\nclass CeleryFilter(object):\n    def filter(self, record):\n        if record.funcName in (\"_log_error\",):\n            return 0\n        else:\n            return 1\n\n\ndef register_exception_tracking(client):\n    dispatch_uid = \"elasticapm-exc-tracking\"\n\n    def process_failure_signal(sender, task_id, exception, args, kwargs, traceback, einfo, **kw):\n        client.capture_exception(\n            extra={\"task_id\": task_id, \"task\": sender, \"args\": args, \"kwargs\": kwargs}, handled=False\n        )\n\n    signals.task_failure.disconnect(process_failure_signal, dispatch_uid=dispatch_uid)\n    signals.task_failure.connect(process_failure_signal, weak=False, dispatch_uid=dispatch_uid)\n    _register_worker_signals(client)\n\n\ndef register_instrumentation(client):\n    def begin_transaction(*args, **kwargs):\n        client.begin_transaction(\"celery\")\n\n    def end_transaction(task_id, task, *args, **kwargs):\n        name = get_name_from_func(task)\n        client.end_transaction(name, kwargs.get(\"state\", \"None\"))\n\n    dispatch_uid = \"elasticapm-tracing-%s\"\n\n    # unregister any existing clients\n    signals.task_prerun.disconnect(begin_transaction, dispatch_uid=dispatch_uid % \"prerun\")\n    signals.task_postrun.disconnect(end_transaction, dispatch_uid=dispatch_uid % \"postrun\")\n\n    # register for this client\n    signals.task_prerun.connect(begin_transaction, dispatch_uid=dispatch_uid % \"prerun\", weak=False)\n    signals.task_postrun.connect(end_transaction, weak=False, dispatch_uid=dispatch_uid % \"postrun\")\n    _register_worker_signals(client)\n\n\ndef _register_worker_signals(client):\n    def worker_startup(*args, **kwargs):\n        client._transport._start_event_processor()\n\n    def worker_shutdown(*args, **kwargs):\n        client.close()\n\n    def connect_worker_process_init(*args, **kwargs):\n        signals.worker_process_init.connect(worker_startup, dispatch_uid=\"elasticapm-start-worker\", weak=False)\n        signals.worker_process_shutdown.connect(worker_shutdown, dispatch_uid=\"elasticapm-shutdown-worker\", weak=False)\n\n    signals.worker_init.connect(\n        connect_worker_process_init, dispatch_uid=\"elasticapm-connect-start-threads\", weak=False\n    )\n", "path": "elasticapm/contrib/celery/__init__.py"}], "after_files": [{"content": "#  BSD 3-Clause License\n#\n#  Copyright (c) 2012, the Sentry Team, see AUTHORS for more details\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\n\nfrom celery import signals\n\nfrom elasticapm.utils import get_name_from_func\n\n\nclass CeleryFilter(object):\n    def filter(self, record):\n        if record.funcName in (\"_log_error\",):\n            return 0\n        else:\n            return 1\n\n\ndef register_exception_tracking(client):\n    dispatch_uid = \"elasticapm-exc-tracking\"\n\n    def process_failure_signal(sender, task_id, exception, args, kwargs, traceback, einfo, **kw):\n        client.capture_exception(\n            extra={\"task_id\": task_id, \"task\": sender, \"args\": args, \"kwargs\": kwargs}, handled=False\n        )\n\n    signals.task_failure.disconnect(process_failure_signal, dispatch_uid=dispatch_uid)\n    signals.task_failure.connect(process_failure_signal, weak=False, dispatch_uid=dispatch_uid)\n    _register_worker_signals(client)\n\n\ndef register_instrumentation(client):\n    def begin_transaction(*args, **kwargs):\n        client.begin_transaction(\"celery\")\n\n    def end_transaction(task_id, task, *args, **kwargs):\n        name = get_name_from_func(task)\n        client.end_transaction(name, kwargs.get(\"state\", \"None\"))\n\n    dispatch_uid = \"elasticapm-tracing-%s\"\n\n    # unregister any existing clients\n    signals.task_prerun.disconnect(begin_transaction, dispatch_uid=dispatch_uid % \"prerun\")\n    signals.task_postrun.disconnect(end_transaction, dispatch_uid=dispatch_uid % \"postrun\")\n\n    # register for this client\n    signals.task_prerun.connect(begin_transaction, dispatch_uid=dispatch_uid % \"prerun\", weak=False)\n    signals.task_postrun.connect(end_transaction, weak=False, dispatch_uid=dispatch_uid % \"postrun\")\n    _register_worker_signals(client)\n\n\ndef _register_worker_signals(client):\n    def worker_shutdown(*args, **kwargs):\n        client.close()\n\n    def connect_worker_process_init(*args, **kwargs):\n        signals.worker_process_shutdown.connect(worker_shutdown, dispatch_uid=\"elasticapm-shutdown-worker\", weak=False)\n\n    signals.worker_init.connect(\n        connect_worker_process_init, dispatch_uid=\"elasticapm-connect-start-threads\", weak=False\n    )\n", "path": "elasticapm/contrib/celery/__init__.py"}]}