We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Reference to undeclared variable: "local variable 'return_code' referenced before assignment"
### Apache Airflow version

2.2.1

### Operating System

Ubuntu 20.04 LTS

### Versions of Apache Airflow Providers

apache-airflow-providers-amazon==2.3.0
apache-airflow-providers-apache-cassandra==2.1.0
apache-airflow-providers-ftp==2.0.1
apache-airflow-providers-google==6.0.0
apache-airflow-providers-http==2.0.1
apache-airflow-providers-imap==2.0.1
apache-airflow-providers-jdbc==2.0.1
apache-airflow-providers-mysql==2.1.1
apache-airflow-providers-postgres==2.3.0
apache-airflow-providers-presto==2.0.1
apache-airflow-providers-slack==4.1.0
apache-airflow-providers-sqlite==2.0.1

### Deployment

Virtualenv installation

### Deployment details

_No response_

### What happened

Incorrect "finally" block invokes "UnboundLocalError: local variable 'return_code' referenced before assignment"
Traceback example:
```python
  File "/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/task/task_runner/standard_task_runner.py", line 88, in _start_by_fork
    self.log.exception(
  File "/usr/lib/python3.8/logging/__init__.py", line 1481, in exception
    self.error(msg, *args, exc_info=exc_info, **kwargs)
  File "/usr/lib/python3.8/logging/__init__.py", line 1475, in error
    self._log(ERROR, msg, args, **kwargs)
  File "/usr/lib/python3.8/logging/__init__.py", line 1589, in _log
    self.handle(record)
  File "/usr/lib/python3.8/logging/__init__.py", line 1599, in handle
    self.callHandlers(record)
  File "/usr/lib/python3.8/logging/__init__.py", line 1661, in callHandlers
    hdlr.handle(record)
  File "/usr/lib/python3.8/logging/__init__.py", line 950, in handle
    rv = self.filter(record)
  File "/usr/lib/python3.8/logging/__init__.py", line 811, in filter
    result = f.filter(record)
  File "/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/utils/log/secrets_masker.py", line 167, in filter
    self._redact_exception_with_context(exc)
  File "/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/utils/log/secrets_masker.py", line 150, in _redact_exception_with_context
    self._redact_exception_with_context(exception.__context__)
  File "/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/utils/log/secrets_masker.py", line 150, in _redact_exception_with_context
    self._redact_exception_with_context(exception.__context__)
  File "/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/utils/log/secrets_masker.py", line 148, in _redact_exception_with_context
    exception.args = (self.redact(v) for v in exception.args)
AttributeError: can't set attribute
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
  File "/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/executors/celery_executor.py", line 121, in _execute_in_fork
    args.func(args)
  File "/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/cli/cli_parser.py", line 48, in command
    return func(*args, **kwargs)
  File "/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/utils/cli.py", line 92, in wrapper
    return f(*args, **kwargs)
  File "/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/cli/commands/task_command.py", line 292, in task_run
    _run_task_by_selected_method(args, dag, ti)
  File "/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/cli/commands/task_command.py", line 105, in _run_task_by_selected_method
    _run_task_by_local_task_job(args, ti)
  File "/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/cli/commands/task_command.py", line 163, in _run_task_by_local_task_job
    run_job.run()
  File "/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/jobs/base_job.py", line 245, in run
    self._execute()
  File "/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/jobs/local_task_job.py", line 103, in _execute
    self.task_runner.start()
  File "/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/task/task_runner/standard_task_runner.py", line 41, in start
    self.process = self._start_by_fork()
  File "/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/task/task_runner/standard_task_runner.py", line 98, in _start_by_fork
    os._exit(return_code)
UnboundLocalError: local variable 'return_code' referenced before assignment
```

Bug location:
https://github.com/apache/airflow/blob/2.2.1/airflow/task/task_runner/standard_task_runner.py#L84-L98

Explanation:
Nested exception triggered when we are trying to log exception, so return_code remains undeclared.

### What you expected to happen

return_code variable should be declared 

### How to reproduce

It is probably hard to reproduce because you need to have exception in task execution as well as exception in logging function.

### Anything else

_No response_

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)


--- 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/task/task_runner/standard_task_runner.py`
Content:
```
1 #
2 # Licensed to the Apache Software Foundation (ASF) under one
3 # or more contributor license agreements.  See the NOTICE file
4 # distributed with this work for additional information
5 # regarding copyright ownership.  The ASF licenses this file
6 # to you under the Apache License, Version 2.0 (the
7 # "License"); you may not use this file except in compliance
8 # with the License.  You may obtain a copy of the License at
9 #
10 #   http://www.apache.org/licenses/LICENSE-2.0
11 #
12 # Unless required by applicable law or agreed to in writing,
13 # software distributed under the License is distributed on an
14 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15 # KIND, either express or implied.  See the License for the
16 # specific language governing permissions and limitations
17 # under the License.
18 """Standard task runner"""
19 import logging
20 import os
21 from typing import Optional
22 
23 import psutil
24 from setproctitle import setproctitle
25 
26 from airflow.settings import CAN_FORK
27 from airflow.task.task_runner.base_task_runner import BaseTaskRunner
28 from airflow.utils.process_utils import reap_process_group
29 
30 
31 class StandardTaskRunner(BaseTaskRunner):
32     """Standard runner for all tasks."""
33 
34     def __init__(self, local_task_job):
35         super().__init__(local_task_job)
36         self._rc = None
37         self.dag = local_task_job.task_instance.task.dag
38 
39     def start(self):
40         if CAN_FORK and not self.run_as_user:
41             self.process = self._start_by_fork()
42         else:
43             self.process = self._start_by_exec()
44 
45     def _start_by_exec(self):
46         subprocess = self.run_command()
47         return psutil.Process(subprocess.pid)
48 
49     def _start_by_fork(self):
50         pid = os.fork()
51         if pid:
52             self.log.info("Started process %d to run task", pid)
53             return psutil.Process(pid)
54         else:
55             import signal
56 
57             from airflow import settings
58             from airflow.cli.cli_parser import get_parser
59             from airflow.sentry import Sentry
60 
61             signal.signal(signal.SIGINT, signal.SIG_DFL)
62             signal.signal(signal.SIGTERM, signal.SIG_DFL)
63             # Start a new process group
64             os.setpgid(0, 0)
65 
66             # Force a new SQLAlchemy session. We can't share open DB handles
67             # between process. The cli code will re-create this as part of its
68             # normal startup
69             settings.engine.pool.dispose()
70             settings.engine.dispose()
71 
72             parser = get_parser()
73             # [1:] - remove "airflow" from the start of the command
74             args = parser.parse_args(self._command[1:])
75 
76             self.log.info('Running: %s', self._command)
77             self.log.info('Job %s: Subtask %s', self._task_instance.job_id, self._task_instance.task_id)
78 
79             proc_title = "airflow task runner: {0.dag_id} {0.task_id} {0.execution_date_or_run_id}"
80             if hasattr(args, "job_id"):
81                 proc_title += " {0.job_id}"
82             setproctitle(proc_title.format(args))
83 
84             try:
85                 args.func(args, dag=self.dag)
86                 return_code = 0
87             except Exception:
88                 self.log.exception(
89                     "Failed to execute job %s for task %s",
90                     self._task_instance.job_id,
91                     self._task_instance.task_id,
92                 )
93                 return_code = 1
94             finally:
95                 # Explicitly flush any pending exception to Sentry if enabled
96                 Sentry.flush()
97                 logging.shutdown()
98                 os._exit(return_code)
99 
100     def return_code(self, timeout: int = 0) -> Optional[int]:
101         # We call this multiple times, but we can only wait on the process once
102         if self._rc is not None or not self.process:
103             return self._rc
104 
105         try:
106             self._rc = self.process.wait(timeout=timeout)
107             self.process = None
108         except psutil.TimeoutExpired:
109             pass
110 
111         return self._rc
112 
113     def terminate(self):
114         if self.process is None:
115             return
116 
117         # Reap the child process - it may already be finished
118         _ = self.return_code(timeout=0)
119 
120         if self.process and self.process.is_running():
121             rcs = reap_process_group(self.process.pid, self.log)
122             self._rc = rcs.get(self.process.pid)
123 
124         self.process = None
125 
126         if self._rc is None:
127             # Something else reaped it before we had a chance, so let's just "guess" at an error code.
128             self._rc = -9
129 
130         if self._rc == -9:
131             # If either we or psutil gives out a -9 return code, it likely means
132             # an OOM happened
133             self.log.error(
134                 'Job %s was killed before it finished (likely due to running out of memory)',
135                 self._task_instance.job_id,
136             )
137
```
--- 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/task/task_runner/standard_task_runner.py b/airflow/task/task_runner/standard_task_runner.py
--- a/airflow/task/task_runner/standard_task_runner.py
+++ b/airflow/task/task_runner/standard_task_runner.py
@@ -85,12 +85,12 @@
                 args.func(args, dag=self.dag)
                 return_code = 0
             except Exception:
+                return_code = 1
                 self.log.exception(
                     "Failed to execute job %s for task %s",
                     self._task_instance.job_id,
                     self._task_instance.task_id,
                 )
-                return_code = 1
             finally:
                 # Explicitly flush any pending exception to Sentry if enabled
                 Sentry.flush()

{"golden_diff": "diff --git a/airflow/task/task_runner/standard_task_runner.py b/airflow/task/task_runner/standard_task_runner.py\n--- a/airflow/task/task_runner/standard_task_runner.py\n+++ b/airflow/task/task_runner/standard_task_runner.py\n@@ -85,12 +85,12 @@\n                 args.func(args, dag=self.dag)\n                 return_code = 0\n             except Exception:\n+                return_code = 1\n                 self.log.exception(\n                     \"Failed to execute job %s for task %s\",\n                     self._task_instance.job_id,\n                     self._task_instance.task_id,\n                 )\n-                return_code = 1\n             finally:\n                 # Explicitly flush any pending exception to Sentry if enabled\n                 Sentry.flush()\n", "issue": "Reference to undeclared variable: \"local variable 'return_code' referenced before assignment\"\n### Apache Airflow version\r\n\r\n2.2.1\r\n\r\n### Operating System\r\n\r\nUbuntu 20.04 LTS\r\n\r\n### Versions of Apache Airflow Providers\r\n\r\napache-airflow-providers-amazon==2.3.0\r\napache-airflow-providers-apache-cassandra==2.1.0\r\napache-airflow-providers-ftp==2.0.1\r\napache-airflow-providers-google==6.0.0\r\napache-airflow-providers-http==2.0.1\r\napache-airflow-providers-imap==2.0.1\r\napache-airflow-providers-jdbc==2.0.1\r\napache-airflow-providers-mysql==2.1.1\r\napache-airflow-providers-postgres==2.3.0\r\napache-airflow-providers-presto==2.0.1\r\napache-airflow-providers-slack==4.1.0\r\napache-airflow-providers-sqlite==2.0.1\r\n\r\n### Deployment\r\n\r\nVirtualenv installation\r\n\r\n### Deployment details\r\n\r\n_No response_\r\n\r\n### What happened\r\n\r\nIncorrect \"finally\" block invokes \"UnboundLocalError: local variable 'return_code' referenced before assignment\"\r\nTraceback example:\r\n```python\r\n  File \"/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/task/task_runner/standard_task_runner.py\", line 88, in _start_by_fork\r\n    self.log.exception(\r\n  File \"/usr/lib/python3.8/logging/__init__.py\", line 1481, in exception\r\n    self.error(msg, *args, exc_info=exc_info, **kwargs)\r\n  File \"/usr/lib/python3.8/logging/__init__.py\", line 1475, in error\r\n    self._log(ERROR, msg, args, **kwargs)\r\n  File \"/usr/lib/python3.8/logging/__init__.py\", line 1589, in _log\r\n    self.handle(record)\r\n  File \"/usr/lib/python3.8/logging/__init__.py\", line 1599, in handle\r\n    self.callHandlers(record)\r\n  File \"/usr/lib/python3.8/logging/__init__.py\", line 1661, in callHandlers\r\n    hdlr.handle(record)\r\n  File \"/usr/lib/python3.8/logging/__init__.py\", line 950, in handle\r\n    rv = self.filter(record)\r\n  File \"/usr/lib/python3.8/logging/__init__.py\", line 811, in filter\r\n    result = f.filter(record)\r\n  File \"/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/utils/log/secrets_masker.py\", line 167, in filter\r\n    self._redact_exception_with_context(exc)\r\n  File \"/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/utils/log/secrets_masker.py\", line 150, in _redact_exception_with_context\r\n    self._redact_exception_with_context(exception.__context__)\r\n  File \"/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/utils/log/secrets_masker.py\", line 150, in _redact_exception_with_context\r\n    self._redact_exception_with_context(exception.__context__)\r\n  File \"/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/utils/log/secrets_masker.py\", line 148, in _redact_exception_with_context\r\n    exception.args = (self.redact(v) for v in exception.args)\r\nAttributeError: can't set attribute\r\nDuring handling of the above exception, another exception occurred:\r\nTraceback (most recent call last):\r\n  File \"/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/executors/celery_executor.py\", line 121, in _execute_in_fork\r\n    args.func(args)\r\n  File \"/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/cli/cli_parser.py\", line 48, in command\r\n    return func(*args, **kwargs)\r\n  File \"/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/utils/cli.py\", line 92, in wrapper\r\n    return f(*args, **kwargs)\r\n  File \"/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/cli/commands/task_command.py\", line 292, in task_run\r\n    _run_task_by_selected_method(args, dag, ti)\r\n  File \"/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/cli/commands/task_command.py\", line 105, in _run_task_by_selected_method\r\n    _run_task_by_local_task_job(args, ti)\r\n  File \"/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/cli/commands/task_command.py\", line 163, in _run_task_by_local_task_job\r\n    run_job.run()\r\n  File \"/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/jobs/base_job.py\", line 245, in run\r\n    self._execute()\r\n  File \"/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/jobs/local_task_job.py\", line 103, in _execute\r\n    self.task_runner.start()\r\n  File \"/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/task/task_runner/standard_task_runner.py\", line 41, in start\r\n    self.process = self._start_by_fork()\r\n  File \"/var/lib/airflow/.venv/lib/python3.8/site-packages/airflow/task/task_runner/standard_task_runner.py\", line 98, in _start_by_fork\r\n    os._exit(return_code)\r\nUnboundLocalError: local variable 'return_code' referenced before assignment\r\n```\r\n\r\nBug location:\r\nhttps://github.com/apache/airflow/blob/2.2.1/airflow/task/task_runner/standard_task_runner.py#L84-L98\r\n\r\nExplanation:\r\nNested exception triggered when we are trying to log exception, so return_code remains undeclared.\r\n\r\n### What you expected to happen\r\n\r\nreturn_code variable should be declared \r\n\r\n### How to reproduce\r\n\r\nIt is probably hard to reproduce because you need to have exception in task execution as well as exception in logging function.\r\n\r\n### Anything else\r\n\r\n_No response_\r\n\r\n### Are you willing to submit PR?\r\n\r\n- [ ] Yes I am willing to submit a PR!\r\n\r\n### Code of Conduct\r\n\r\n- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)\r\n\n", "before_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\"\"\"Standard task runner\"\"\"\nimport logging\nimport os\nfrom typing import Optional\n\nimport psutil\nfrom setproctitle import setproctitle\n\nfrom airflow.settings import CAN_FORK\nfrom airflow.task.task_runner.base_task_runner import BaseTaskRunner\nfrom airflow.utils.process_utils import reap_process_group\n\n\nclass StandardTaskRunner(BaseTaskRunner):\n    \"\"\"Standard runner for all tasks.\"\"\"\n\n    def __init__(self, local_task_job):\n        super().__init__(local_task_job)\n        self._rc = None\n        self.dag = local_task_job.task_instance.task.dag\n\n    def start(self):\n        if CAN_FORK and not self.run_as_user:\n            self.process = self._start_by_fork()\n        else:\n            self.process = self._start_by_exec()\n\n    def _start_by_exec(self):\n        subprocess = self.run_command()\n        return psutil.Process(subprocess.pid)\n\n    def _start_by_fork(self):\n        pid = os.fork()\n        if pid:\n            self.log.info(\"Started process %d to run task\", pid)\n            return psutil.Process(pid)\n        else:\n            import signal\n\n            from airflow import settings\n            from airflow.cli.cli_parser import get_parser\n            from airflow.sentry import Sentry\n\n            signal.signal(signal.SIGINT, signal.SIG_DFL)\n            signal.signal(signal.SIGTERM, signal.SIG_DFL)\n            # Start a new process group\n            os.setpgid(0, 0)\n\n            # Force a new SQLAlchemy session. We can't share open DB handles\n            # between process. The cli code will re-create this as part of its\n            # normal startup\n            settings.engine.pool.dispose()\n            settings.engine.dispose()\n\n            parser = get_parser()\n            # [1:] - remove \"airflow\" from the start of the command\n            args = parser.parse_args(self._command[1:])\n\n            self.log.info('Running: %s', self._command)\n            self.log.info('Job %s: Subtask %s', self._task_instance.job_id, self._task_instance.task_id)\n\n            proc_title = \"airflow task runner: {0.dag_id} {0.task_id} {0.execution_date_or_run_id}\"\n            if hasattr(args, \"job_id\"):\n                proc_title += \" {0.job_id}\"\n            setproctitle(proc_title.format(args))\n\n            try:\n                args.func(args, dag=self.dag)\n                return_code = 0\n            except Exception:\n                self.log.exception(\n                    \"Failed to execute job %s for task %s\",\n                    self._task_instance.job_id,\n                    self._task_instance.task_id,\n                )\n                return_code = 1\n            finally:\n                # Explicitly flush any pending exception to Sentry if enabled\n                Sentry.flush()\n                logging.shutdown()\n                os._exit(return_code)\n\n    def return_code(self, timeout: int = 0) -> Optional[int]:\n        # We call this multiple times, but we can only wait on the process once\n        if self._rc is not None or not self.process:\n            return self._rc\n\n        try:\n            self._rc = self.process.wait(timeout=timeout)\n            self.process = None\n        except psutil.TimeoutExpired:\n            pass\n\n        return self._rc\n\n    def terminate(self):\n        if self.process is None:\n            return\n\n        # Reap the child process - it may already be finished\n        _ = self.return_code(timeout=0)\n\n        if self.process and self.process.is_running():\n            rcs = reap_process_group(self.process.pid, self.log)\n            self._rc = rcs.get(self.process.pid)\n\n        self.process = None\n\n        if self._rc is None:\n            # Something else reaped it before we had a chance, so let's just \"guess\" at an error code.\n            self._rc = -9\n\n        if self._rc == -9:\n            # If either we or psutil gives out a -9 return code, it likely means\n            # an OOM happened\n            self.log.error(\n                'Job %s was killed before it finished (likely due to running out of memory)',\n                self._task_instance.job_id,\n            )\n", "path": "airflow/task/task_runner/standard_task_runner.py"}], "after_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\"\"\"Standard task runner\"\"\"\nimport logging\nimport os\nfrom typing import Optional\n\nimport psutil\nfrom setproctitle import setproctitle\n\nfrom airflow.settings import CAN_FORK\nfrom airflow.task.task_runner.base_task_runner import BaseTaskRunner\nfrom airflow.utils.process_utils import reap_process_group\n\n\nclass StandardTaskRunner(BaseTaskRunner):\n    \"\"\"Standard runner for all tasks.\"\"\"\n\n    def __init__(self, local_task_job):\n        super().__init__(local_task_job)\n        self._rc = None\n        self.dag = local_task_job.task_instance.task.dag\n\n    def start(self):\n        if CAN_FORK and not self.run_as_user:\n            self.process = self._start_by_fork()\n        else:\n            self.process = self._start_by_exec()\n\n    def _start_by_exec(self):\n        subprocess = self.run_command()\n        return psutil.Process(subprocess.pid)\n\n    def _start_by_fork(self):\n        pid = os.fork()\n        if pid:\n            self.log.info(\"Started process %d to run task\", pid)\n            return psutil.Process(pid)\n        else:\n            import signal\n\n            from airflow import settings\n            from airflow.cli.cli_parser import get_parser\n            from airflow.sentry import Sentry\n\n            signal.signal(signal.SIGINT, signal.SIG_DFL)\n            signal.signal(signal.SIGTERM, signal.SIG_DFL)\n            # Start a new process group\n            os.setpgid(0, 0)\n\n            # Force a new SQLAlchemy session. We can't share open DB handles\n            # between process. The cli code will re-create this as part of its\n            # normal startup\n            settings.engine.pool.dispose()\n            settings.engine.dispose()\n\n            parser = get_parser()\n            # [1:] - remove \"airflow\" from the start of the command\n            args = parser.parse_args(self._command[1:])\n\n            self.log.info('Running: %s', self._command)\n            self.log.info('Job %s: Subtask %s', self._task_instance.job_id, self._task_instance.task_id)\n\n            proc_title = \"airflow task runner: {0.dag_id} {0.task_id} {0.execution_date_or_run_id}\"\n            if hasattr(args, \"job_id\"):\n                proc_title += \" {0.job_id}\"\n            setproctitle(proc_title.format(args))\n\n            try:\n                args.func(args, dag=self.dag)\n                return_code = 0\n            except Exception:\n                return_code = 1\n                self.log.exception(\n                    \"Failed to execute job %s for task %s\",\n                    self._task_instance.job_id,\n                    self._task_instance.task_id,\n                )\n            finally:\n                # Explicitly flush any pending exception to Sentry if enabled\n                Sentry.flush()\n                logging.shutdown()\n                os._exit(return_code)\n\n    def return_code(self, timeout: int = 0) -> Optional[int]:\n        # We call this multiple times, but we can only wait on the process once\n        if self._rc is not None or not self.process:\n            return self._rc\n\n        try:\n            self._rc = self.process.wait(timeout=timeout)\n            self.process = None\n        except psutil.TimeoutExpired:\n            pass\n\n        return self._rc\n\n    def terminate(self):\n        if self.process is None:\n            return\n\n        # Reap the child process - it may already be finished\n        _ = self.return_code(timeout=0)\n\n        if self.process and self.process.is_running():\n            rcs = reap_process_group(self.process.pid, self.log)\n            self._rc = rcs.get(self.process.pid)\n\n        self.process = None\n\n        if self._rc is None:\n            # Something else reaped it before we had a chance, so let's just \"guess\" at an error code.\n            self._rc = -9\n\n        if self._rc == -9:\n            # If either we or psutil gives out a -9 return code, it likely means\n            # an OOM happened\n            self.log.error(\n                'Job %s was killed before it finished (likely due to running out of memory)',\n                self._task_instance.job_id,\n            )\n", "path": "airflow/task/task_runner/standard_task_runner.py"}]}