We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Airflow 2.7.1 can not start Scheduler & trigger
### Apache Airflow version

2.7.1

### What happened

After upgrade from 2.6.0 to 2.7.1 (try pip uninstall apache-airflow, and clear dir airflow - remove airflow.cfg), I can start scheduler & trigger with daemon. 
I try start with command, it can start, but logout console it killed.
I try: airflow scheduler or airflow triggerer :done but kill when logout console
airflow scheduler --daemon && airflow triggerer --daemon: fail, can not start scheduler & triggerer (2.6.0 run ok). but start deamon with webserver & celery worker is fine

Help me

### What you think should happen instead

_No response_

### How to reproduce

1. run airflow 2.6.0 fine on ubuntu server 22.04.3 lts
2. install airflow 2.7.1 
3. can not start daemon triggerer & scheduler

### Operating System

ubuntu server 22.04.3 LTS

### Versions of Apache Airflow Providers

_No response_

### Deployment

Virtualenv installation

### Deployment details

_No response_

### Anything else

_No response_

### Are you willing to submit PR?

- [X] 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/cli/commands/triggerer_command.py`
Content:
```
1 # Licensed to the Apache Software Foundation (ASF) under one
2 # or more contributor license agreements.  See the NOTICE file
3 # distributed with this work for additional information
4 # regarding copyright ownership.  The ASF licenses this file
5 # to you under the Apache License, Version 2.0 (the
6 # "License"); you may not use this file except in compliance
7 # with the License.  You may obtain a copy of the License at
8 #
9 #   http://www.apache.org/licenses/LICENSE-2.0
10 #
11 # Unless required by applicable law or agreed to in writing,
12 # software distributed under the License is distributed on an
13 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14 # KIND, either express or implied.  See the License for the
15 # specific language governing permissions and limitations
16 # under the License.
17 """Triggerer command."""
18 from __future__ import annotations
19 
20 import signal
21 from contextlib import contextmanager
22 from functools import partial
23 from multiprocessing import Process
24 from typing import Generator
25 
26 import daemon
27 from daemon.pidfile import TimeoutPIDLockFile
28 
29 from airflow import settings
30 from airflow.configuration import conf
31 from airflow.jobs.job import Job, run_job
32 from airflow.jobs.triggerer_job_runner import TriggererJobRunner
33 from airflow.utils import cli as cli_utils
34 from airflow.utils.cli import setup_locations, setup_logging, sigint_handler, sigquit_handler
35 from airflow.utils.providers_configuration_loader import providers_configuration_loaded
36 from airflow.utils.serve_logs import serve_logs
37 
38 
39 @contextmanager
40 def _serve_logs(skip_serve_logs: bool = False) -> Generator[None, None, None]:
41     """Start serve_logs sub-process."""
42     sub_proc = None
43     if skip_serve_logs is False:
44         port = conf.getint("logging", "trigger_log_server_port", fallback=8794)
45         sub_proc = Process(target=partial(serve_logs, port=port))
46         sub_proc.start()
47     try:
48         yield
49     finally:
50         if sub_proc:
51             sub_proc.terminate()
52 
53 
54 @cli_utils.action_cli
55 @providers_configuration_loaded
56 def triggerer(args):
57     """Start Airflow Triggerer."""
58     settings.MASK_SECRETS_IN_LOGS = True
59     print(settings.HEADER)
60     triggerer_heartrate = conf.getfloat("triggerer", "JOB_HEARTBEAT_SEC")
61     triggerer_job_runner = TriggererJobRunner(job=Job(heartrate=triggerer_heartrate), capacity=args.capacity)
62 
63     if args.daemon:
64         pid, stdout, stderr, log_file = setup_locations(
65             "triggerer", args.pid, args.stdout, args.stderr, args.log_file
66         )
67         handle = setup_logging(log_file)
68         with open(stdout, "a") as stdout_handle, open(stderr, "a") as stderr_handle:
69             stdout_handle.truncate(0)
70             stderr_handle.truncate(0)
71 
72             daemon_context = daemon.DaemonContext(
73                 pidfile=TimeoutPIDLockFile(pid, -1),
74                 files_preserve=[handle],
75                 stdout=stdout_handle,
76                 stderr=stderr_handle,
77                 umask=int(settings.DAEMON_UMASK, 8),
78             )
79             with daemon_context, _serve_logs(args.skip_serve_logs):
80                 run_job(job=triggerer_job_runner.job, execute_callable=triggerer_job_runner._execute)
81     else:
82         signal.signal(signal.SIGINT, sigint_handler)
83         signal.signal(signal.SIGTERM, sigint_handler)
84         signal.signal(signal.SIGQUIT, sigquit_handler)
85         with _serve_logs(args.skip_serve_logs):
86             run_job(job=triggerer_job_runner.job, execute_callable=triggerer_job_runner._execute)
87
```
--- 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/cli/commands/triggerer_command.py b/airflow/cli/commands/triggerer_command.py
--- a/airflow/cli/commands/triggerer_command.py
+++ b/airflow/cli/commands/triggerer_command.py
@@ -58,7 +58,6 @@
     settings.MASK_SECRETS_IN_LOGS = True
     print(settings.HEADER)
     triggerer_heartrate = conf.getfloat("triggerer", "JOB_HEARTBEAT_SEC")
-    triggerer_job_runner = TriggererJobRunner(job=Job(heartrate=triggerer_heartrate), capacity=args.capacity)
 
     if args.daemon:
         pid, stdout, stderr, log_file = setup_locations(
@@ -77,10 +76,16 @@
                 umask=int(settings.DAEMON_UMASK, 8),
             )
             with daemon_context, _serve_logs(args.skip_serve_logs):
+                triggerer_job_runner = TriggererJobRunner(
+                    job=Job(heartrate=triggerer_heartrate), capacity=args.capacity
+                )
                 run_job(job=triggerer_job_runner.job, execute_callable=triggerer_job_runner._execute)
     else:
         signal.signal(signal.SIGINT, sigint_handler)
         signal.signal(signal.SIGTERM, sigint_handler)
         signal.signal(signal.SIGQUIT, sigquit_handler)
         with _serve_logs(args.skip_serve_logs):
+            triggerer_job_runner = TriggererJobRunner(
+                job=Job(heartrate=triggerer_heartrate), capacity=args.capacity
+            )
             run_job(job=triggerer_job_runner.job, execute_callable=triggerer_job_runner._execute)

{"golden_diff": "diff --git a/airflow/cli/commands/triggerer_command.py b/airflow/cli/commands/triggerer_command.py\n--- a/airflow/cli/commands/triggerer_command.py\n+++ b/airflow/cli/commands/triggerer_command.py\n@@ -58,7 +58,6 @@\n     settings.MASK_SECRETS_IN_LOGS = True\n     print(settings.HEADER)\n     triggerer_heartrate = conf.getfloat(\"triggerer\", \"JOB_HEARTBEAT_SEC\")\n-    triggerer_job_runner = TriggererJobRunner(job=Job(heartrate=triggerer_heartrate), capacity=args.capacity)\n \n     if args.daemon:\n         pid, stdout, stderr, log_file = setup_locations(\n@@ -77,10 +76,16 @@\n                 umask=int(settings.DAEMON_UMASK, 8),\n             )\n             with daemon_context, _serve_logs(args.skip_serve_logs):\n+                triggerer_job_runner = TriggererJobRunner(\n+                    job=Job(heartrate=triggerer_heartrate), capacity=args.capacity\n+                )\n                 run_job(job=triggerer_job_runner.job, execute_callable=triggerer_job_runner._execute)\n     else:\n         signal.signal(signal.SIGINT, sigint_handler)\n         signal.signal(signal.SIGTERM, sigint_handler)\n         signal.signal(signal.SIGQUIT, sigquit_handler)\n         with _serve_logs(args.skip_serve_logs):\n+            triggerer_job_runner = TriggererJobRunner(\n+                job=Job(heartrate=triggerer_heartrate), capacity=args.capacity\n+            )\n             run_job(job=triggerer_job_runner.job, execute_callable=triggerer_job_runner._execute)\n", "issue": "Airflow 2.7.1 can not start Scheduler & trigger\n### Apache Airflow version\n\n2.7.1\n\n### What happened\n\nAfter upgrade from 2.6.0 to 2.7.1 (try pip uninstall apache-airflow, and clear dir airflow - remove airflow.cfg), I can start scheduler & trigger with daemon. \r\nI try start with command, it can start, but logout console it killed.\r\nI try: airflow scheduler or airflow triggerer :done but kill when logout console\r\nairflow scheduler --daemon && airflow triggerer --daemon: fail, can not start scheduler & triggerer (2.6.0 run ok). but start deamon with webserver & celery worker is fine\r\n\r\nHelp me\n\n### What you think should happen instead\n\n_No response_\n\n### How to reproduce\n\n1. run airflow 2.6.0 fine on ubuntu server 22.04.3 lts\r\n2. install airflow 2.7.1 \r\n3. can not start daemon triggerer & scheduler\n\n### Operating System\n\nubuntu server 22.04.3 LTS\n\n### Versions of Apache Airflow Providers\n\n_No response_\n\n### Deployment\n\nVirtualenv installation\n\n### Deployment details\n\n_No response_\n\n### Anything else\n\n_No response_\n\n### Are you willing to submit PR?\n\n- [X] Yes I am willing to submit a PR!\n\n### Code of Conduct\n\n- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)\n\n", "before_files": [{"content": "# 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\"\"\"Triggerer command.\"\"\"\nfrom __future__ import annotations\n\nimport signal\nfrom contextlib import contextmanager\nfrom functools import partial\nfrom multiprocessing import Process\nfrom typing import Generator\n\nimport daemon\nfrom daemon.pidfile import TimeoutPIDLockFile\n\nfrom airflow import settings\nfrom airflow.configuration import conf\nfrom airflow.jobs.job import Job, run_job\nfrom airflow.jobs.triggerer_job_runner import TriggererJobRunner\nfrom airflow.utils import cli as cli_utils\nfrom airflow.utils.cli import setup_locations, setup_logging, sigint_handler, sigquit_handler\nfrom airflow.utils.providers_configuration_loader import providers_configuration_loaded\nfrom airflow.utils.serve_logs import serve_logs\n\n\n@contextmanager\ndef _serve_logs(skip_serve_logs: bool = False) -> Generator[None, None, None]:\n    \"\"\"Start serve_logs sub-process.\"\"\"\n    sub_proc = None\n    if skip_serve_logs is False:\n        port = conf.getint(\"logging\", \"trigger_log_server_port\", fallback=8794)\n        sub_proc = Process(target=partial(serve_logs, port=port))\n        sub_proc.start()\n    try:\n        yield\n    finally:\n        if sub_proc:\n            sub_proc.terminate()\n\n\n@cli_utils.action_cli\n@providers_configuration_loaded\ndef triggerer(args):\n    \"\"\"Start Airflow Triggerer.\"\"\"\n    settings.MASK_SECRETS_IN_LOGS = True\n    print(settings.HEADER)\n    triggerer_heartrate = conf.getfloat(\"triggerer\", \"JOB_HEARTBEAT_SEC\")\n    triggerer_job_runner = TriggererJobRunner(job=Job(heartrate=triggerer_heartrate), capacity=args.capacity)\n\n    if args.daemon:\n        pid, stdout, stderr, log_file = setup_locations(\n            \"triggerer\", args.pid, args.stdout, args.stderr, args.log_file\n        )\n        handle = setup_logging(log_file)\n        with open(stdout, \"a\") as stdout_handle, open(stderr, \"a\") as stderr_handle:\n            stdout_handle.truncate(0)\n            stderr_handle.truncate(0)\n\n            daemon_context = daemon.DaemonContext(\n                pidfile=TimeoutPIDLockFile(pid, -1),\n                files_preserve=[handle],\n                stdout=stdout_handle,\n                stderr=stderr_handle,\n                umask=int(settings.DAEMON_UMASK, 8),\n            )\n            with daemon_context, _serve_logs(args.skip_serve_logs):\n                run_job(job=triggerer_job_runner.job, execute_callable=triggerer_job_runner._execute)\n    else:\n        signal.signal(signal.SIGINT, sigint_handler)\n        signal.signal(signal.SIGTERM, sigint_handler)\n        signal.signal(signal.SIGQUIT, sigquit_handler)\n        with _serve_logs(args.skip_serve_logs):\n            run_job(job=triggerer_job_runner.job, execute_callable=triggerer_job_runner._execute)\n", "path": "airflow/cli/commands/triggerer_command.py"}], "after_files": [{"content": "# 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\"\"\"Triggerer command.\"\"\"\nfrom __future__ import annotations\n\nimport signal\nfrom contextlib import contextmanager\nfrom functools import partial\nfrom multiprocessing import Process\nfrom typing import Generator\n\nimport daemon\nfrom daemon.pidfile import TimeoutPIDLockFile\n\nfrom airflow import settings\nfrom airflow.configuration import conf\nfrom airflow.jobs.job import Job, run_job\nfrom airflow.jobs.triggerer_job_runner import TriggererJobRunner\nfrom airflow.utils import cli as cli_utils\nfrom airflow.utils.cli import setup_locations, setup_logging, sigint_handler, sigquit_handler\nfrom airflow.utils.providers_configuration_loader import providers_configuration_loaded\nfrom airflow.utils.serve_logs import serve_logs\n\n\n@contextmanager\ndef _serve_logs(skip_serve_logs: bool = False) -> Generator[None, None, None]:\n    \"\"\"Start serve_logs sub-process.\"\"\"\n    sub_proc = None\n    if skip_serve_logs is False:\n        port = conf.getint(\"logging\", \"trigger_log_server_port\", fallback=8794)\n        sub_proc = Process(target=partial(serve_logs, port=port))\n        sub_proc.start()\n    try:\n        yield\n    finally:\n        if sub_proc:\n            sub_proc.terminate()\n\n\n@cli_utils.action_cli\n@providers_configuration_loaded\ndef triggerer(args):\n    \"\"\"Start Airflow Triggerer.\"\"\"\n    settings.MASK_SECRETS_IN_LOGS = True\n    print(settings.HEADER)\n    triggerer_heartrate = conf.getfloat(\"triggerer\", \"JOB_HEARTBEAT_SEC\")\n\n    if args.daemon:\n        pid, stdout, stderr, log_file = setup_locations(\n            \"triggerer\", args.pid, args.stdout, args.stderr, args.log_file\n        )\n        handle = setup_logging(log_file)\n        with open(stdout, \"a\") as stdout_handle, open(stderr, \"a\") as stderr_handle:\n            stdout_handle.truncate(0)\n            stderr_handle.truncate(0)\n\n            daemon_context = daemon.DaemonContext(\n                pidfile=TimeoutPIDLockFile(pid, -1),\n                files_preserve=[handle],\n                stdout=stdout_handle,\n                stderr=stderr_handle,\n                umask=int(settings.DAEMON_UMASK, 8),\n            )\n            with daemon_context, _serve_logs(args.skip_serve_logs):\n                triggerer_job_runner = TriggererJobRunner(\n                    job=Job(heartrate=triggerer_heartrate), capacity=args.capacity\n                )\n                run_job(job=triggerer_job_runner.job, execute_callable=triggerer_job_runner._execute)\n    else:\n        signal.signal(signal.SIGINT, sigint_handler)\n        signal.signal(signal.SIGTERM, sigint_handler)\n        signal.signal(signal.SIGQUIT, sigquit_handler)\n        with _serve_logs(args.skip_serve_logs):\n            triggerer_job_runner = TriggererJobRunner(\n                job=Job(heartrate=triggerer_heartrate), capacity=args.capacity\n            )\n            run_job(job=triggerer_job_runner.job, execute_callable=triggerer_job_runner._execute)\n", "path": "airflow/cli/commands/triggerer_command.py"}]}