diff --git a/airflow/providers/http/hooks/http.py b/airflow/providers/http/hooks/http.py
index 76ded12276682..00231eb64f9c6 100644
--- a/airflow/providers/http/hooks/http.py
+++ b/airflow/providers/http/hooks/http.py
@@ -45,7 +45,7 @@
         "requests.auth.HTTPBasicAuth",
         "requests.auth.HTTPProxyAuth",
         "requests.auth.HTTPDigestAuth",
-        "aiohttp.BasicAuth"
+        "aiohttp.BasicAuth",
     }
 )
 
@@ -153,7 +153,6 @@ def load_connection_settings(self, *, headers: dict[Any, Any] | None = None) ->
                 self.base_url += f":{conn.port}"
 
             conn_extra: dict = self._parse_extra(conn_extra=conn.extra_dejson)
-            print(conn_extra)
             auth_args: list[str | None] = [conn.login, conn.password]
             auth_kwargs: dict[str, Any] = conn_extra["auth_kwargs"]
             auth_type: Any = (
@@ -490,7 +489,7 @@ async def run(
             ``aiohttp.ClientSession().get(json=obj)``.
         """
         extra_options = extra_options or {}
-        headers, auth = await sync_to_async(self.load_connection_settings)(headers)
+        headers, auth = await sync_to_async(self.load_connection_settings)(headers=headers)
 
         base_url = (self.base_url or "").rstrip("/")
         endpoint = (endpoint or "").lstrip("/")
diff --git a/tests/providers/http/hooks/test_http.py b/tests/providers/http/hooks/test_http.py
index add81d8cd324a..62f729f1db4ef 100644
--- a/tests/providers/http/hooks/test_http.py
+++ b/tests/providers/http/hooks/test_http.py
@@ -362,6 +362,32 @@ def test_connection_with_extra_auth_type_and_no_credentials(self, auth, mock_get
         HttpHook().get_conn({})
         auth.assert_called_once()
 
+    @mock.patch("airflow.providers.http.hooks.http.HttpHook.get_connection")
+    @mock.patch("tests.providers.http.hooks.test_http.CustomAuthBase.__init__")
+    def test_connection_with_string_headers_and_auth_kwargs(self, auth, mock_get_connection):
+        """When passed via the UI, the 'headers' and 'auth_kwargs' fields' data is
+        saved as string.
+        """
+        auth.return_value = None
+        conn = Connection(
+            conn_id="http_default",
+            conn_type="http",
+            login="username",
+            password="pass",
+            extra=r"""
+                {"auth_kwargs": "{\r\n    \"endpoint\": \"http://localhost\"\r\n}",
+                "headers": "{\r\n    \"some\": \"headers\"\r\n}"}
+                """,
+        )
+        mock_get_connection.return_value = conn
+
+        hook = HttpHook(auth_type=CustomAuthBase)
+        session = hook.get_conn({})
+
+        auth.assert_called_once_with("username", "pass", endpoint="http://localhost")
+        assert "auth_kwargs" not in session.headers
+        assert "some" in session.headers
+
     @pytest.mark.parametrize("method", ["GET", "POST"])
     def test_json_request(self, method, requests_mock):
         obj1 = {"a": 1, "b": "abc", "c": [1, 2, {"d": 10}]}
@@ -602,7 +628,7 @@ async def test_async_post_request_with_error_code(self, aioresponse):
     async def test_async_request_uses_connection_extra(self, aioresponse):
         """Test api call asynchronously with a connection that has extra field."""
 
-        connection_extra = {"bearer": "test"}
+        connection_extra = {"bearer": "test", "some": "header"}
 
         aioresponse.post(
             "http://test:8080/v1/test",