kaxil · Aug 27, 2021 · Jun 11, 2021 · Jun 15, 2021 · Jun 23, 2021 · Jun 29, 2021
diff --git a/airflow/configuration.py b/airflow/configuration.py
@@ -81,10 +81,18 @@ def run_command(command):
 
 def _get_config_value_from_secret_backend(config_key):
     """Get Config option values from Secret Backend"""
-    secrets_client = get_custom_secret_backend()
-    if not secrets_client:
-        return None
-    return secrets_client.get_config(config_key)
+    try:
+        secrets_client = get_custom_secret_backend()
+        if not secrets_client:
+            return None
+        return secrets_client.get_config(config_key)
+    except Exception as e:  # pylint: disable=broad-except
+        raise AirflowConfigException(
+            'Cannot retrieve config from alternative secrets backend. '
+            'Make sure it is configured properly and that the Backend '
+            'is accessible.\n'
+            f'{e}'
+        )
 
 
 def _default_config_file_path(file_name: str):

diff --git a/airflow/models/connection.py b/airflow/models/connection.py
@@ -17,6 +17,7 @@
 # under the License.
 
 import json
+import logging
 import warnings
 from json import JSONDecodeError
 from typing import Dict, Optional, Union
@@ -35,6 +36,8 @@
 from airflow.utils.log.secrets_masker import mask_secret
 from airflow.utils.module_loading import import_string
 
+log = logging.getLogger(__name__)
+
 
 def parse_netloc_to_hostname(*args, **kwargs):
     """This method is deprecated."""
@@ -389,7 +392,15 @@ def get_connection_from_secrets(cls, conn_id: str) -> 'Connection':
         :return: connection
         """
         for secrets_backend in ensure_secrets_loaded():
-            conn = secrets_backend.get_connection(conn_id=conn_id)
-            if conn:
-                return conn
+            try:
+                conn = secrets_backend.get_connection(conn_id=conn_id)
+                if conn:
+                    return conn
+            except Exception:  # pylint: disable=broad-except
+                log.exception(
+                    'Unable to retrieve connection from secrets backend (%s). '
+                    'Checking subsequent secrets backend.',
+                    type(secrets_backend).__name__,
+                )
+
         raise AirflowNotFoundException(f"The conn_id `{conn_id}` isn't defined")
diff --git a/airflow/models/variable.py b/airflow/models/variable.py
@@ -33,7 +33,7 @@
 from airflow.utils.log.secrets_masker import mask_secret
 from airflow.utils.session import provide_session
 
-log = logging.getLogger()
+log = logging.getLogger(__name__)
 
 
 class Variable(Base, LoggingMixin):
@@ -201,7 +201,14 @@ def get_variable_from_secrets(key: str) -> Optional[str]:
         :return: Variable Value
         """
         for secrets_backend in ensure_secrets_loaded():
-            var_val = secrets_backend.get_variable(key=key)
-            if var_val is not None:
-                return var_val
+            try:
+                var_val = secrets_backend.get_variable(key=key)
+                if var_val is not None:
+                    return var_val
+            except Exception:  # pylint: disable=broad-except
+                log.exception(
+                    'Unable to retrieve variable from secrets backend (%s). '
+                    'Checking subsequent secrets backend.',
+                    type(secrets_backend).__name__,
+                )
         return None
diff --git a/tests/core/test_configuration.py b/tests/core/test_configuration.py
@@ -251,6 +251,43 @@ def test_config_from_secret_backend(self, mock_hvac):
 
         assert 'sqlite:////Users/airflow/airflow/airflow.db' == test_conf.get('test', 'sql_alchemy_conn')
 
+    @mock.patch("airflow.providers.hashicorp._internal_client.vault_client.hvac")
+    @conf_vars(
+        {
+            ("secrets", "backend"): "airflow.providers.hashicorp.secrets.vault.VaultBackend",
+            ("secrets", "backend_kwargs"): '{"url": "http://127.0.0.1:8200", "token": "token"}',
+        }
+    )
+    def test_config_raise_exception_from_secret_backend_connection_error(self, mock_hvac):
+        """Get Config Value from a Secret Backend"""
+
+        mock_client = mock.MagicMock()
+        # mock_client.side_effect = AirflowConfigException
+        mock_hvac.Client.return_value = mock_client
+        mock_client.secrets.kv.v2.read_secret_version.return_value = Exception
+
+        test_config = '''[test]
+sql_alchemy_conn_secret = sql_alchemy_conn
+'''
+        test_config_default = '''[test]
+sql_alchemy_conn = airflow
+'''
+        test_conf = AirflowConfigParser(default_config=parameterized_config(test_config_default))
+        test_conf.read_string(test_config)
+        test_conf.sensitive_config_values = test_conf.sensitive_config_values | {
+            ('test', 'sql_alchemy_conn'),
+        }
+
+        with pytest.raises(
+            AirflowConfigException,
+            match=re.escape(
+                'Cannot retrieve config from alternative secrets backend. '
+                'Make sure it is configured properly and that the Backend '
+                'is accessible.'
+            ),
+        ):
+            test_conf.get('test', 'sql_alchemy_conn')
+
     def test_getboolean(self):
         """Test AirflowConfigParser.getboolean"""
         test_config = """