We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
db+ string in result backend but not metadata secret
### Official Helm Chart version

1.1.0 (latest released)

### Apache Airflow version

2.1.3 (latest released)

### Kubernetes Version

1.21

### Helm Chart configuration

  data:
    metadataSecretName: "airflow-metadata"
    resultBackendSecretName: "airflow-result-backend"

### Docker Image customisations

_No response_

### What happened

If we only supply 1 secret with 
```
connection: postgresql://airflow:password@postgres.rds:5432/airflow?sslmode=disable
```
To use for both metadata and resultBackendConnection then we end up with a connection error because
resultBackendConnection expects the string to be formatted like 
```
connection: db+postgresql://airflow:password@postgres.rds:5432/airflow?sslmode=disable
```
from what i can tell

### What you expected to happen

I'd expect to be able to use the same secret for both using the same format if they are using the same connection. 

### How to reproduce

Make a secret structured like above to look like the metadataConnection auto-generated secret.
use that same secret for the result backend.
deploy.


### Anything else

Occurs always. 
To get around currently we make 2 secrets one with just the db+ prepended. 

### 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/config_templates/default_celery.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 """Default celery configuration."""
19 import logging
20 import ssl
21 
22 from airflow.configuration import conf
23 from airflow.exceptions import AirflowConfigException, AirflowException
24 
25 
26 def _broker_supports_visibility_timeout(url):
27     return url.startswith("redis://") or url.startswith("sqs://")
28 
29 
30 log = logging.getLogger(__name__)
31 
32 broker_url = conf.get('celery', 'BROKER_URL')
33 
34 broker_transport_options = conf.getsection('celery_broker_transport_options') or {}
35 if 'visibility_timeout' not in broker_transport_options:
36     if _broker_supports_visibility_timeout(broker_url):
37         broker_transport_options['visibility_timeout'] = 21600
38 
39 DEFAULT_CELERY_CONFIG = {
40     'accept_content': ['json'],
41     'event_serializer': 'json',
42     'worker_prefetch_multiplier': conf.getint('celery', 'worker_prefetch_multiplier'),
43     'task_acks_late': True,
44     'task_default_queue': conf.get('operators', 'DEFAULT_QUEUE'),
45     'task_default_exchange': conf.get('operators', 'DEFAULT_QUEUE'),
46     'task_track_started': conf.getboolean('celery', 'task_track_started'),
47     'broker_url': broker_url,
48     'broker_transport_options': broker_transport_options,
49     'result_backend': conf.get('celery', 'RESULT_BACKEND'),
50     'worker_concurrency': conf.getint('celery', 'WORKER_CONCURRENCY'),
51     'worker_enable_remote_control': conf.getboolean('celery', 'worker_enable_remote_control'),
52 }
53 
54 celery_ssl_active = False
55 try:
56     celery_ssl_active = conf.getboolean('celery', 'SSL_ACTIVE')
57 except AirflowConfigException:
58     log.warning("Celery Executor will run without SSL")
59 
60 try:
61     if celery_ssl_active:
62         if broker_url and 'amqp://' in broker_url:
63             broker_use_ssl = {
64                 'keyfile': conf.get('celery', 'SSL_KEY'),
65                 'certfile': conf.get('celery', 'SSL_CERT'),
66                 'ca_certs': conf.get('celery', 'SSL_CACERT'),
67                 'cert_reqs': ssl.CERT_REQUIRED,
68             }
69         elif broker_url and 'redis://' in broker_url:
70             broker_use_ssl = {
71                 'ssl_keyfile': conf.get('celery', 'SSL_KEY'),
72                 'ssl_certfile': conf.get('celery', 'SSL_CERT'),
73                 'ssl_ca_certs': conf.get('celery', 'SSL_CACERT'),
74                 'ssl_cert_reqs': ssl.CERT_REQUIRED,
75             }
76         else:
77             raise AirflowException(
78                 'The broker you configured does not support SSL_ACTIVE to be True. '
79                 'Please use RabbitMQ or Redis if you would like to use SSL for broker.'
80             )
81 
82         DEFAULT_CELERY_CONFIG['broker_use_ssl'] = broker_use_ssl
83 except AirflowConfigException:
84     raise AirflowException(
85         'AirflowConfigException: SSL_ACTIVE is True, '
86         'please ensure SSL_KEY, '
87         'SSL_CERT and SSL_CACERT are set'
88     )
89 except Exception as e:
90     raise AirflowException(
91         f'Exception: There was an unknown Celery SSL Error. Please ensure you want to use SSL and/or have '
92         f'all necessary certs and key ({e}).'
93     )
94 
95 result_backend = str(DEFAULT_CELERY_CONFIG['result_backend'])
96 if 'amqp://' in result_backend or 'redis://' in result_backend or 'rpc://' in result_backend:
97     log.warning(
98         "You have configured a result_backend of %s, it is highly recommended "
99         "to use an alternative result_backend (i.e. a database).",
100         result_backend,
101     )
102
```
--- 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/config_templates/default_celery.py b/airflow/config_templates/default_celery.py
--- a/airflow/config_templates/default_celery.py
+++ b/airflow/config_templates/default_celery.py
@@ -36,6 +36,12 @@
     if _broker_supports_visibility_timeout(broker_url):
         broker_transport_options['visibility_timeout'] = 21600
 
+if conf.has_option("celery", 'RESULT_BACKEND'):
+    result_backend = conf.get_mandatory_value('celery', 'RESULT_BACKEND')
+else:
+    log.debug("Value for celery result_backend not found. Using sql_alchemy_conn with db+ prefix.")
+    result_backend = f'db+{conf.get("database", "SQL_ALCHEMY_CONN")}'
+
 DEFAULT_CELERY_CONFIG = {
     'accept_content': ['json'],
     'event_serializer': 'json',
@@ -46,7 +52,7 @@
     'task_track_started': conf.getboolean('celery', 'task_track_started'),
     'broker_url': broker_url,
     'broker_transport_options': broker_transport_options,
-    'result_backend': conf.get('celery', 'RESULT_BACKEND'),
+    'result_backend': result_backend,
     'worker_concurrency': conf.getint('celery', 'WORKER_CONCURRENCY'),
     'worker_enable_remote_control': conf.getboolean('celery', 'worker_enable_remote_control'),
 }
@@ -92,7 +98,6 @@
         f'all necessary certs and key ({e}).'
     )
 
-result_backend = str(DEFAULT_CELERY_CONFIG['result_backend'])
 if 'amqp://' in result_backend or 'redis://' in result_backend or 'rpc://' in result_backend:
     log.warning(
         "You have configured a result_backend of %s, it is highly recommended "

{"golden_diff": "diff --git a/airflow/config_templates/default_celery.py b/airflow/config_templates/default_celery.py\n--- a/airflow/config_templates/default_celery.py\n+++ b/airflow/config_templates/default_celery.py\n@@ -36,6 +36,12 @@\n     if _broker_supports_visibility_timeout(broker_url):\n         broker_transport_options['visibility_timeout'] = 21600\n \n+if conf.has_option(\"celery\", 'RESULT_BACKEND'):\n+    result_backend = conf.get_mandatory_value('celery', 'RESULT_BACKEND')\n+else:\n+    log.debug(\"Value for celery result_backend not found. Using sql_alchemy_conn with db+ prefix.\")\n+    result_backend = f'db+{conf.get(\"database\", \"SQL_ALCHEMY_CONN\")}'\n+\n DEFAULT_CELERY_CONFIG = {\n     'accept_content': ['json'],\n     'event_serializer': 'json',\n@@ -46,7 +52,7 @@\n     'task_track_started': conf.getboolean('celery', 'task_track_started'),\n     'broker_url': broker_url,\n     'broker_transport_options': broker_transport_options,\n-    'result_backend': conf.get('celery', 'RESULT_BACKEND'),\n+    'result_backend': result_backend,\n     'worker_concurrency': conf.getint('celery', 'WORKER_CONCURRENCY'),\n     'worker_enable_remote_control': conf.getboolean('celery', 'worker_enable_remote_control'),\n }\n@@ -92,7 +98,6 @@\n         f'all necessary certs and key ({e}).'\n     )\n \n-result_backend = str(DEFAULT_CELERY_CONFIG['result_backend'])\n if 'amqp://' in result_backend or 'redis://' in result_backend or 'rpc://' in result_backend:\n     log.warning(\n         \"You have configured a result_backend of %s, it is highly recommended \"\n", "issue": "db+ string in result backend but not metadata secret\n### Official Helm Chart version\n\n1.1.0 (latest released)\n\n### Apache Airflow version\n\n2.1.3 (latest released)\n\n### Kubernetes Version\n\n1.21\n\n### Helm Chart configuration\n\n  data:\r\n    metadataSecretName: \"airflow-metadata\"\r\n    resultBackendSecretName: \"airflow-result-backend\"\n\n### Docker Image customisations\n\n_No response_\n\n### What happened\n\nIf we only supply 1 secret with \r\n```\r\nconnection: postgresql://airflow:password@postgres.rds:5432/airflow?sslmode=disable\r\n```\r\nTo use for both metadata and resultBackendConnection then we end up with a connection error because\r\nresultBackendConnection expects the string to be formatted like \r\n```\r\nconnection: db+postgresql://airflow:password@postgres.rds:5432/airflow?sslmode=disable\r\n```\r\nfrom what i can tell\n\n### What you expected to happen\n\nI'd expect to be able to use the same secret for both using the same format if they are using the same connection. \n\n### How to reproduce\n\nMake a secret structured like above to look like the metadataConnection auto-generated secret.\r\nuse that same secret for the result backend.\r\ndeploy.\r\n\n\n### Anything else\n\nOccurs always. \r\nTo get around currently we make 2 secrets one with just the db+ prepended. \n\n### Are you willing to submit PR?\n\n- [ ] 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": "#\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\"\"\"Default celery configuration.\"\"\"\nimport logging\nimport ssl\n\nfrom airflow.configuration import conf\nfrom airflow.exceptions import AirflowConfigException, AirflowException\n\n\ndef _broker_supports_visibility_timeout(url):\n    return url.startswith(\"redis://\") or url.startswith(\"sqs://\")\n\n\nlog = logging.getLogger(__name__)\n\nbroker_url = conf.get('celery', 'BROKER_URL')\n\nbroker_transport_options = conf.getsection('celery_broker_transport_options') or {}\nif 'visibility_timeout' not in broker_transport_options:\n    if _broker_supports_visibility_timeout(broker_url):\n        broker_transport_options['visibility_timeout'] = 21600\n\nDEFAULT_CELERY_CONFIG = {\n    'accept_content': ['json'],\n    'event_serializer': 'json',\n    'worker_prefetch_multiplier': conf.getint('celery', 'worker_prefetch_multiplier'),\n    'task_acks_late': True,\n    'task_default_queue': conf.get('operators', 'DEFAULT_QUEUE'),\n    'task_default_exchange': conf.get('operators', 'DEFAULT_QUEUE'),\n    'task_track_started': conf.getboolean('celery', 'task_track_started'),\n    'broker_url': broker_url,\n    'broker_transport_options': broker_transport_options,\n    'result_backend': conf.get('celery', 'RESULT_BACKEND'),\n    'worker_concurrency': conf.getint('celery', 'WORKER_CONCURRENCY'),\n    'worker_enable_remote_control': conf.getboolean('celery', 'worker_enable_remote_control'),\n}\n\ncelery_ssl_active = False\ntry:\n    celery_ssl_active = conf.getboolean('celery', 'SSL_ACTIVE')\nexcept AirflowConfigException:\n    log.warning(\"Celery Executor will run without SSL\")\n\ntry:\n    if celery_ssl_active:\n        if broker_url and 'amqp://' in broker_url:\n            broker_use_ssl = {\n                'keyfile': conf.get('celery', 'SSL_KEY'),\n                'certfile': conf.get('celery', 'SSL_CERT'),\n                'ca_certs': conf.get('celery', 'SSL_CACERT'),\n                'cert_reqs': ssl.CERT_REQUIRED,\n            }\n        elif broker_url and 'redis://' in broker_url:\n            broker_use_ssl = {\n                'ssl_keyfile': conf.get('celery', 'SSL_KEY'),\n                'ssl_certfile': conf.get('celery', 'SSL_CERT'),\n                'ssl_ca_certs': conf.get('celery', 'SSL_CACERT'),\n                'ssl_cert_reqs': ssl.CERT_REQUIRED,\n            }\n        else:\n            raise AirflowException(\n                'The broker you configured does not support SSL_ACTIVE to be True. '\n                'Please use RabbitMQ or Redis if you would like to use SSL for broker.'\n            )\n\n        DEFAULT_CELERY_CONFIG['broker_use_ssl'] = broker_use_ssl\nexcept AirflowConfigException:\n    raise AirflowException(\n        'AirflowConfigException: SSL_ACTIVE is True, '\n        'please ensure SSL_KEY, '\n        'SSL_CERT and SSL_CACERT are set'\n    )\nexcept Exception as e:\n    raise AirflowException(\n        f'Exception: There was an unknown Celery SSL Error. Please ensure you want to use SSL and/or have '\n        f'all necessary certs and key ({e}).'\n    )\n\nresult_backend = str(DEFAULT_CELERY_CONFIG['result_backend'])\nif 'amqp://' in result_backend or 'redis://' in result_backend or 'rpc://' in result_backend:\n    log.warning(\n        \"You have configured a result_backend of %s, it is highly recommended \"\n        \"to use an alternative result_backend (i.e. a database).\",\n        result_backend,\n    )\n", "path": "airflow/config_templates/default_celery.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\"\"\"Default celery configuration.\"\"\"\nimport logging\nimport ssl\n\nfrom airflow.configuration import conf\nfrom airflow.exceptions import AirflowConfigException, AirflowException\n\n\ndef _broker_supports_visibility_timeout(url):\n    return url.startswith(\"redis://\") or url.startswith(\"sqs://\")\n\n\nlog = logging.getLogger(__name__)\n\nbroker_url = conf.get('celery', 'BROKER_URL')\n\nbroker_transport_options = conf.getsection('celery_broker_transport_options') or {}\nif 'visibility_timeout' not in broker_transport_options:\n    if _broker_supports_visibility_timeout(broker_url):\n        broker_transport_options['visibility_timeout'] = 21600\n\nif conf.has_option(\"celery\", 'RESULT_BACKEND'):\n    result_backend = conf.get_mandatory_value('celery', 'RESULT_BACKEND')\nelse:\n    log.debug(\"Value for celery result_backend not found. Using sql_alchemy_conn with db+ prefix.\")\n    result_backend = f'db+{conf.get(\"database\", \"SQL_ALCHEMY_CONN\")}'\n\nDEFAULT_CELERY_CONFIG = {\n    'accept_content': ['json'],\n    'event_serializer': 'json',\n    'worker_prefetch_multiplier': conf.getint('celery', 'worker_prefetch_multiplier'),\n    'task_acks_late': True,\n    'task_default_queue': conf.get('operators', 'DEFAULT_QUEUE'),\n    'task_default_exchange': conf.get('operators', 'DEFAULT_QUEUE'),\n    'task_track_started': conf.getboolean('celery', 'task_track_started'),\n    'broker_url': broker_url,\n    'broker_transport_options': broker_transport_options,\n    'result_backend': result_backend,\n    'worker_concurrency': conf.getint('celery', 'WORKER_CONCURRENCY'),\n    'worker_enable_remote_control': conf.getboolean('celery', 'worker_enable_remote_control'),\n}\n\ncelery_ssl_active = False\ntry:\n    celery_ssl_active = conf.getboolean('celery', 'SSL_ACTIVE')\nexcept AirflowConfigException:\n    log.warning(\"Celery Executor will run without SSL\")\n\ntry:\n    if celery_ssl_active:\n        if broker_url and 'amqp://' in broker_url:\n            broker_use_ssl = {\n                'keyfile': conf.get('celery', 'SSL_KEY'),\n                'certfile': conf.get('celery', 'SSL_CERT'),\n                'ca_certs': conf.get('celery', 'SSL_CACERT'),\n                'cert_reqs': ssl.CERT_REQUIRED,\n            }\n        elif broker_url and 'redis://' in broker_url:\n            broker_use_ssl = {\n                'ssl_keyfile': conf.get('celery', 'SSL_KEY'),\n                'ssl_certfile': conf.get('celery', 'SSL_CERT'),\n                'ssl_ca_certs': conf.get('celery', 'SSL_CACERT'),\n                'ssl_cert_reqs': ssl.CERT_REQUIRED,\n            }\n        else:\n            raise AirflowException(\n                'The broker you configured does not support SSL_ACTIVE to be True. '\n                'Please use RabbitMQ or Redis if you would like to use SSL for broker.'\n            )\n\n        DEFAULT_CELERY_CONFIG['broker_use_ssl'] = broker_use_ssl\nexcept AirflowConfigException:\n    raise AirflowException(\n        'AirflowConfigException: SSL_ACTIVE is True, '\n        'please ensure SSL_KEY, '\n        'SSL_CERT and SSL_CACERT are set'\n    )\nexcept Exception as e:\n    raise AirflowException(\n        f'Exception: There was an unknown Celery SSL Error. Please ensure you want to use SSL and/or have '\n        f'all necessary certs and key ({e}).'\n    )\n\nif 'amqp://' in result_backend or 'redis://' in result_backend or 'rpc://' in result_backend:\n    log.warning(\n        \"You have configured a result_backend of %s, it is highly recommended \"\n        \"to use an alternative result_backend (i.e. a database).\",\n        result_backend,\n    )\n", "path": "airflow/config_templates/default_celery.py"}]}