_prepare_credentials: connector fix for fsspec>=2022.12

simplify _prepare credentials, revert some of the
changes related to iterative/dvc#7414

Author: dtrifiro

dvc_http/__init__.py

@@ -1,6 +1,6 @@
 import threading
 from getpass import getpass
-from typing import BinaryIO, Optional, Union
+from typing import Any, BinaryIO, Dict, Optional, Union
 
 from dvc_objects.fs.base import AnyFSPath, FileSystem
 from dvc_objects.fs.callbacks import DEFAULT_CALLBACK, Callback
@@ -40,7 +40,6 @@ class HTTPFileSystem(FileSystem):
 
     def _prepare_credentials(self, **config):
         import aiohttp
-        from fsspec.asyn import fsspec_loop
 
         credentials = {}
         client_kwargs = credentials.setdefault("client_kwargs", {})
@@ -74,20 +73,11 @@ def _prepare_credentials(self, **config):
                     f"Auth method {auth_method!r} is not supported."
                 )
 
-        # Force cleanup of closed SSL transports.
-        # https://github.com/iterative/dvc/issues/7414
-        connector_kwargs = {"enable_cleanup_closed": True}
-
         if "ssl_verify" in config:
-            connector_kwargs.update(ssl=make_context(config["ssl_verify"]))
-
-        with fsspec_loop():
-            client_kwargs["connector"] = aiohttp.TCPConnector(
-                **connector_kwargs
-            )
-        # The connector should not be owned by aiohttp.ClientSession since
-        # it is closed by fsspec (HTTPFileSystem.close_session)
-        client_kwargs["connector_owner"] = False
+            client_kwargs["connector_kwargs"] = {
+                "ssl": make_context(config["ssl_verify"])
+            }
+
         client_kwargs["connect_timeout"] = config.get(
             "connect_timeout", self.REQUEST_TIMEOUT
         )
@@ -110,6 +100,7 @@ async def get_client(
         connect_timeout: Optional[float],
         sock_connect_timeout: Optional[float],
         sock_read_timeout: Optional[float],
+        connector_kwargs: Optional[Dict[str, Any]] = None,
         **kwargs,
     ):
         import aiohttp
@@ -136,6 +127,9 @@ async def get_client(
             sock_read=sock_read_timeout,
         )
 
+        if connector_kwargs:
+            kwargs["connector"] = aiohttp.TCPConnector(**connector_kwargs)
+
         return ReadOnlyRetryClient(**kwargs)
 
     @cached_property