We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[BUG] FutureWarning: pyarrow.hdfs.connect is deprecated
### Willingness to contribute

Yes. I can contribute a fix for this bug independently.

### MLflow version

2.2.2

### System information

- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: CentOS 7
- **Python version**: 3.9.6
- **pyarrow version**: 5.0.0


### Describe the problem

When loading a model from the MLflow Model Registry (backend store is HDFS) I get the following warning:

```
legobricks/lib/python3.9/site-packages/mlflow/store/artifact/hdfs_artifact_repo.py:188: FutureWarning: pyarrow.hdfs.connect is deprecated as of 2.0.0, please use pyarrow.fs.HadoopFileSystem instead.
  connected = pa.hdfs.connect(
```

### Steps to reproduce the bug

1. Create a conda env with mlflow and pyarrow installed
2. Start Mlflow with hdfs as backend store
3. Train a model and save it
4. Store the trained model in the model registry
5. Load model from the registry

### Code to generate data required to reproduce the bug

_No response_

### Is the console panel in DevTools showing errors relevant to the bug?

_No response_

### Does the network panel in DevTools contain failed requests relevant to the bug?

_No response_

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `mlflow/store/artifact/hdfs_artifact_repo.py`
Content:
```
1 import os
2 import posixpath
3 import tempfile
4 import urllib.parse
5 from contextlib import contextmanager
6 
7 from mlflow.entities import FileInfo
8 from mlflow.environment_variables import (
9     MLFLOW_KERBEROS_TICKET_CACHE,
10     MLFLOW_KERBEROS_USER,
11     MLFLOW_PYARROW_EXTRA_CONF,
12 )
13 from mlflow.exceptions import MlflowException
14 from mlflow.store.artifact.artifact_repo import ArtifactRepository
15 from mlflow.utils.file_utils import mkdir, relative_path_to_artifact_path
16 
17 
18 class HdfsArtifactRepository(ArtifactRepository):
19     """
20     Stores artifacts on HDFS.
21 
22     This repository is used with URIs of the form ``hdfs:/<path>``. The repository can only be used
23     together with the RestStore.
24     """
25 
26     def __init__(self, artifact_uri):
27         self.scheme, self.host, self.port, self.path = _resolve_connection_params(artifact_uri)
28         super().__init__(artifact_uri)
29 
30     def log_artifact(self, local_file, artifact_path=None):
31         """
32             Log artifact in hdfs.
33         :param local_file: source file path
34         :param artifact_path: when specified will attempt to write under artifact_uri/artifact_path
35         """
36         hdfs_base_path = _resolve_base_path(self.path, artifact_path)
37 
38         with hdfs_system(scheme=self.scheme, host=self.host, port=self.port) as hdfs:
39             _, file_name = os.path.split(local_file)
40             destination = posixpath.join(hdfs_base_path, file_name)
41             with open(local_file, "rb") as f:
42                 hdfs.upload(destination, f)
43 
44     def log_artifacts(self, local_dir, artifact_path=None):
45         """
46             Log artifacts in hdfs.
47             Missing remote sub-directories will be created if needed.
48         :param local_dir: source dir path
49         :param artifact_path: when specified will attempt to write under artifact_uri/artifact_path
50         """
51         hdfs_base_path = _resolve_base_path(self.path, artifact_path)
52 
53         with hdfs_system(scheme=self.scheme, host=self.host, port=self.port) as hdfs:
54             if not hdfs.exists(hdfs_base_path):
55                 hdfs.mkdir(hdfs_base_path)
56 
57             for subdir_path, _, files in os.walk(local_dir):
58                 relative_path = _relative_path_local(local_dir, subdir_path)
59 
60                 hdfs_subdir_path = (
61                     posixpath.join(hdfs_base_path, relative_path)
62                     if relative_path
63                     else hdfs_base_path
64                 )
65 
66                 if not hdfs.exists(hdfs_subdir_path):
67                     hdfs.mkdir(hdfs_subdir_path)
68 
69                 for each_file in files:
70                     source = os.path.join(subdir_path, each_file)
71                     destination = posixpath.join(hdfs_subdir_path, each_file)
72                     with open(source, "rb") as f:
73                         hdfs.upload(destination, f)
74 
75     def list_artifacts(self, path=None):
76         """
77         Lists files and directories under artifacts directory for the current run_id.
78         (self.path contains the base path - hdfs:/some/path/run_id/artifacts)
79 
80         :param path: Relative source path. Possible subdirectory existing under
81                      hdfs:/some/path/run_id/artifacts
82         :return: List of FileInfos under given path
83         """
84         hdfs_base_path = _resolve_base_path(self.path, path)
85 
86         with hdfs_system(scheme=self.scheme, host=self.host, port=self.port) as hdfs:
87             paths = []
88             if hdfs.exists(hdfs_base_path):
89                 for file_detail in hdfs.ls(hdfs_base_path, detail=True):
90                     file_name = file_detail.get("name")
91 
92                     # file_name is hdfs_base_path and not a child of that path
93                     if file_name == hdfs_base_path:
94                         continue
95 
96                     # Strip off anything that comes before the artifact root e.g. hdfs://name
97                     offset = file_name.index(self.path)
98                     rel_path = _relative_path_remote(self.path, file_name[offset:])
99                     is_dir = file_detail.get("kind") == "directory"
100                     size = file_detail.get("size")
101                     paths.append(FileInfo(rel_path, is_dir=is_dir, file_size=size))
102             return sorted(paths, key=lambda f: paths)
103 
104     def _walk_path(self, hdfs, hdfs_path):
105         if hdfs.exists(hdfs_path):
106             if hdfs.isdir(hdfs_path):
107                 for subdir, _, files in hdfs.walk(hdfs_path):
108                     if subdir != hdfs_path:
109                         yield subdir, hdfs.isdir(subdir), hdfs.info(subdir).get("size")
110                     for f in files:
111                         file_path = posixpath.join(subdir, f)
112                         yield file_path, hdfs.isdir(file_path), hdfs.info(file_path).get("size")
113             else:
114                 yield hdfs_path, False, hdfs.info(hdfs_path).get("size")
115 
116     def download_artifacts(self, artifact_path, dst_path=None):
117         """
118         Download an artifact file or directory to a local directory/file if applicable, and
119         return a local path for it.
120         The caller is responsible for managing the lifecycle of the downloaded artifacts.
121 
122         (self.path contains the base path - hdfs:/some/path/run_id/artifacts)
123 
124         :param artifact_path: Relative source path to the desired artifacts file or directory.
125         :param dst_path: Absolute path of the local filesystem destination directory to which
126                          to download the specified artifacts. This directory must already
127                          exist. If unspecified, the artifacts will be downloaded to a new,
128                          uniquely-named
129                          directory on the local filesystem.
130 
131         :return: Absolute path of the local filesystem location containing the downloaded
132         artifacts - file/directory.
133         """
134 
135         hdfs_base_path = _resolve_base_path(self.path, artifact_path)
136         if dst_path and os.path.exists(dst_path):
137             local_dir = os.path.abspath(dst_path)
138         else:
139             local_dir = _tmp_dir(dst_path)
140 
141         with hdfs_system(scheme=self.scheme, host=self.host, port=self.port) as hdfs:
142             if not hdfs.isdir(hdfs_base_path):
143                 local_path = os.path.join(local_dir, os.path.normpath(artifact_path))
144                 _download_hdfs_file(hdfs, hdfs_base_path, local_path)
145                 return local_path
146 
147             for path, is_dir, _ in self._walk_path(hdfs, hdfs_base_path):
148                 relative_path = _relative_path_remote(hdfs_base_path, path)
149                 local_path = os.path.join(local_dir, relative_path) if relative_path else local_dir
150 
151                 if is_dir:
152                     mkdir(local_path)
153                 else:
154                     _download_hdfs_file(hdfs, path, local_path)
155             return local_dir
156 
157     def _download_file(self, remote_file_path, local_path):
158         raise MlflowException("This is not implemented. Should never be called.")
159 
160     def delete_artifacts(self, artifact_path=None):
161         path = posixpath.join(self.path, artifact_path) if artifact_path else self.path
162         with hdfs_system(scheme=self.scheme, host=self.host, port=self.port) as hdfs:
163             hdfs.delete(path, recursive=True)
164 
165 
166 @contextmanager
167 def hdfs_system(scheme, host, port):
168     """
169         hdfs system context - Attempt to establish the connection to hdfs
170         and yields HadoopFileSystem
171 
172     :param scheme: scheme or use hdfs:// as default
173     :param host: hostname or when relaying on the core-site.xml config use 'default'
174     :param port: port or when relaying on the core-site.xml config use 0
175     """
176     import pyarrow as pa
177 
178     kerb_ticket = MLFLOW_KERBEROS_TICKET_CACHE.get()
179     kerberos_user = MLFLOW_KERBEROS_USER.get()
180     extra_conf = _parse_extra_conf(MLFLOW_PYARROW_EXTRA_CONF.get())
181 
182     if host:
183         host = scheme + "://" + host
184     else:
185         host = "default"
186 
187     connected = pa.hdfs.connect(
188         host=host,
189         port=port or 0,
190         user=kerberos_user,
191         kerb_ticket=kerb_ticket,
192         extra_conf=extra_conf,
193     )
194     yield connected
195     connected.close()
196 
197 
198 def _resolve_connection_params(artifact_uri):
199     parsed = urllib.parse.urlparse(artifact_uri)
200 
201     return parsed.scheme, parsed.hostname, parsed.port, parsed.path
202 
203 
204 def _resolve_base_path(path, artifact_path):
205     if path == artifact_path:
206         return path
207     if artifact_path:
208         return posixpath.join(path, artifact_path)
209     return path
210 
211 
212 def _relative_path(base_dir, subdir_path, path_module):
213     relative_path = path_module.relpath(subdir_path, base_dir)
214     return relative_path if relative_path != "." else None
215 
216 
217 def _relative_path_local(base_dir, subdir_path):
218     rel_path = _relative_path(base_dir, subdir_path, os.path)
219     return relative_path_to_artifact_path(rel_path) if rel_path is not None else None
220 
221 
222 def _relative_path_remote(base_dir, subdir_path):
223     return _relative_path(base_dir, subdir_path, posixpath)
224 
225 
226 def _tmp_dir(local_path):
227     return os.path.abspath(tempfile.mkdtemp(dir=local_path))
228 
229 
230 def _download_hdfs_file(hdfs, remote_file_path, local_file_path):
231     # Ensure all required directories exist. Without doing this nested files can't be downloaded.
232     dirs = os.path.dirname(local_file_path)
233     if not os.path.exists(dirs):
234         os.makedirs(dirs)
235     with open(local_file_path, "wb") as f:
236         f.write(hdfs.open(remote_file_path, "rb").read())
237 
238 
239 def _parse_extra_conf(extra_conf):
240     if extra_conf:
241 
242         def as_pair(config):
243             key, val = config.split("=")
244             return key, val
245 
246         list_of_key_val = [as_pair(conf) for conf in extra_conf.split(",")]
247         return dict(list_of_key_val)
248     return None
249
```
--- 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/mlflow/store/artifact/hdfs_artifact_repo.py b/mlflow/store/artifact/hdfs_artifact_repo.py
--- a/mlflow/store/artifact/hdfs_artifact_repo.py
+++ b/mlflow/store/artifact/hdfs_artifact_repo.py
@@ -4,6 +4,8 @@
 import urllib.parse
 from contextlib import contextmanager
 
+import packaging.version
+
 from mlflow.entities import FileInfo
 from mlflow.environment_variables import (
     MLFLOW_KERBEROS_TICKET_CACHE,
@@ -173,7 +175,7 @@
     :param host: hostname or when relaying on the core-site.xml config use 'default'
     :param port: port or when relaying on the core-site.xml config use 0
     """
-    import pyarrow as pa
+    import pyarrow
 
     kerb_ticket = MLFLOW_KERBEROS_TICKET_CACHE.get()
     kerberos_user = MLFLOW_KERBEROS_USER.get()
@@ -183,14 +185,22 @@
         host = scheme + "://" + host
     else:
         host = "default"
-
-    connected = pa.hdfs.connect(
-        host=host,
-        port=port or 0,
-        user=kerberos_user,
-        kerb_ticket=kerb_ticket,
-        extra_conf=extra_conf,
-    )
+    if packaging.version.parse(pyarrow.__version__) < packaging.version.parse("2.0.0"):
+        connected = pyarrow.fs.HadoopFileSystem(
+            host=host,
+            port=port or 0,
+            user=kerberos_user,
+            kerb_ticket=kerb_ticket,
+            extra_conf=extra_conf,
+        )
+    else:
+        connected = pyarrow.hdfs.connect(
+            host=host,
+            port=port or 0,
+            user=kerberos_user,
+            kerb_ticket=kerb_ticket,
+            extra_conf=extra_conf,
+        )
     yield connected
     connected.close()

{"golden_diff": "diff --git a/mlflow/store/artifact/hdfs_artifact_repo.py b/mlflow/store/artifact/hdfs_artifact_repo.py\n--- a/mlflow/store/artifact/hdfs_artifact_repo.py\n+++ b/mlflow/store/artifact/hdfs_artifact_repo.py\n@@ -4,6 +4,8 @@\n import urllib.parse\n from contextlib import contextmanager\n \n+import packaging.version\n+\n from mlflow.entities import FileInfo\n from mlflow.environment_variables import (\n     MLFLOW_KERBEROS_TICKET_CACHE,\n@@ -173,7 +175,7 @@\n     :param host: hostname or when relaying on the core-site.xml config use 'default'\n     :param port: port or when relaying on the core-site.xml config use 0\n     \"\"\"\n-    import pyarrow as pa\n+    import pyarrow\n \n     kerb_ticket = MLFLOW_KERBEROS_TICKET_CACHE.get()\n     kerberos_user = MLFLOW_KERBEROS_USER.get()\n@@ -183,14 +185,22 @@\n         host = scheme + \"://\" + host\n     else:\n         host = \"default\"\n-\n-    connected = pa.hdfs.connect(\n-        host=host,\n-        port=port or 0,\n-        user=kerberos_user,\n-        kerb_ticket=kerb_ticket,\n-        extra_conf=extra_conf,\n-    )\n+    if packaging.version.parse(pyarrow.__version__) < packaging.version.parse(\"2.0.0\"):\n+        connected = pyarrow.fs.HadoopFileSystem(\n+            host=host,\n+            port=port or 0,\n+            user=kerberos_user,\n+            kerb_ticket=kerb_ticket,\n+            extra_conf=extra_conf,\n+        )\n+    else:\n+        connected = pyarrow.hdfs.connect(\n+            host=host,\n+            port=port or 0,\n+            user=kerberos_user,\n+            kerb_ticket=kerb_ticket,\n+            extra_conf=extra_conf,\n+        )\n     yield connected\n     connected.close()\n", "issue": "[BUG] FutureWarning: pyarrow.hdfs.connect is deprecated\n### Willingness to contribute\n\nYes. I can contribute a fix for this bug independently.\n\n### MLflow version\n\n2.2.2\n\n### System information\n\n- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: CentOS 7\r\n- **Python version**: 3.9.6\r\n- **pyarrow version**: 5.0.0\r\n\n\n### Describe the problem\n\nWhen loading a model from the MLflow Model Registry (backend store is HDFS) I get the following warning:\r\n\r\n```\r\nlegobricks/lib/python3.9/site-packages/mlflow/store/artifact/hdfs_artifact_repo.py:188: FutureWarning: pyarrow.hdfs.connect is deprecated as of 2.0.0, please use pyarrow.fs.HadoopFileSystem instead.\r\n  connected = pa.hdfs.connect(\r\n```\n\n### Steps to reproduce the bug\n\n1. Create a conda env with mlflow and pyarrow installed\r\n2. Start Mlflow with hdfs as backend store\r\n3. Train a model and save it\r\n4. Store the trained model in the model registry\r\n5. Load model from the registry\n\n### Code to generate data required to reproduce the bug\n\n_No response_\n\n### Is the console panel in DevTools showing errors relevant to the bug?\n\n_No response_\n\n### Does the network panel in DevTools contain failed requests relevant to the bug?\n\n_No response_\n", "before_files": [{"content": "import os\nimport posixpath\nimport tempfile\nimport urllib.parse\nfrom contextlib import contextmanager\n\nfrom mlflow.entities import FileInfo\nfrom mlflow.environment_variables import (\n    MLFLOW_KERBEROS_TICKET_CACHE,\n    MLFLOW_KERBEROS_USER,\n    MLFLOW_PYARROW_EXTRA_CONF,\n)\nfrom mlflow.exceptions import MlflowException\nfrom mlflow.store.artifact.artifact_repo import ArtifactRepository\nfrom mlflow.utils.file_utils import mkdir, relative_path_to_artifact_path\n\n\nclass HdfsArtifactRepository(ArtifactRepository):\n    \"\"\"\n    Stores artifacts on HDFS.\n\n    This repository is used with URIs of the form ``hdfs:/<path>``. The repository can only be used\n    together with the RestStore.\n    \"\"\"\n\n    def __init__(self, artifact_uri):\n        self.scheme, self.host, self.port, self.path = _resolve_connection_params(artifact_uri)\n        super().__init__(artifact_uri)\n\n    def log_artifact(self, local_file, artifact_path=None):\n        \"\"\"\n            Log artifact in hdfs.\n        :param local_file: source file path\n        :param artifact_path: when specified will attempt to write under artifact_uri/artifact_path\n        \"\"\"\n        hdfs_base_path = _resolve_base_path(self.path, artifact_path)\n\n        with hdfs_system(scheme=self.scheme, host=self.host, port=self.port) as hdfs:\n            _, file_name = os.path.split(local_file)\n            destination = posixpath.join(hdfs_base_path, file_name)\n            with open(local_file, \"rb\") as f:\n                hdfs.upload(destination, f)\n\n    def log_artifacts(self, local_dir, artifact_path=None):\n        \"\"\"\n            Log artifacts in hdfs.\n            Missing remote sub-directories will be created if needed.\n        :param local_dir: source dir path\n        :param artifact_path: when specified will attempt to write under artifact_uri/artifact_path\n        \"\"\"\n        hdfs_base_path = _resolve_base_path(self.path, artifact_path)\n\n        with hdfs_system(scheme=self.scheme, host=self.host, port=self.port) as hdfs:\n            if not hdfs.exists(hdfs_base_path):\n                hdfs.mkdir(hdfs_base_path)\n\n            for subdir_path, _, files in os.walk(local_dir):\n                relative_path = _relative_path_local(local_dir, subdir_path)\n\n                hdfs_subdir_path = (\n                    posixpath.join(hdfs_base_path, relative_path)\n                    if relative_path\n                    else hdfs_base_path\n                )\n\n                if not hdfs.exists(hdfs_subdir_path):\n                    hdfs.mkdir(hdfs_subdir_path)\n\n                for each_file in files:\n                    source = os.path.join(subdir_path, each_file)\n                    destination = posixpath.join(hdfs_subdir_path, each_file)\n                    with open(source, \"rb\") as f:\n                        hdfs.upload(destination, f)\n\n    def list_artifacts(self, path=None):\n        \"\"\"\n        Lists files and directories under artifacts directory for the current run_id.\n        (self.path contains the base path - hdfs:/some/path/run_id/artifacts)\n\n        :param path: Relative source path. Possible subdirectory existing under\n                     hdfs:/some/path/run_id/artifacts\n        :return: List of FileInfos under given path\n        \"\"\"\n        hdfs_base_path = _resolve_base_path(self.path, path)\n\n        with hdfs_system(scheme=self.scheme, host=self.host, port=self.port) as hdfs:\n            paths = []\n            if hdfs.exists(hdfs_base_path):\n                for file_detail in hdfs.ls(hdfs_base_path, detail=True):\n                    file_name = file_detail.get(\"name\")\n\n                    # file_name is hdfs_base_path and not a child of that path\n                    if file_name == hdfs_base_path:\n                        continue\n\n                    # Strip off anything that comes before the artifact root e.g. hdfs://name\n                    offset = file_name.index(self.path)\n                    rel_path = _relative_path_remote(self.path, file_name[offset:])\n                    is_dir = file_detail.get(\"kind\") == \"directory\"\n                    size = file_detail.get(\"size\")\n                    paths.append(FileInfo(rel_path, is_dir=is_dir, file_size=size))\n            return sorted(paths, key=lambda f: paths)\n\n    def _walk_path(self, hdfs, hdfs_path):\n        if hdfs.exists(hdfs_path):\n            if hdfs.isdir(hdfs_path):\n                for subdir, _, files in hdfs.walk(hdfs_path):\n                    if subdir != hdfs_path:\n                        yield subdir, hdfs.isdir(subdir), hdfs.info(subdir).get(\"size\")\n                    for f in files:\n                        file_path = posixpath.join(subdir, f)\n                        yield file_path, hdfs.isdir(file_path), hdfs.info(file_path).get(\"size\")\n            else:\n                yield hdfs_path, False, hdfs.info(hdfs_path).get(\"size\")\n\n    def download_artifacts(self, artifact_path, dst_path=None):\n        \"\"\"\n        Download an artifact file or directory to a local directory/file if applicable, and\n        return a local path for it.\n        The caller is responsible for managing the lifecycle of the downloaded artifacts.\n\n        (self.path contains the base path - hdfs:/some/path/run_id/artifacts)\n\n        :param artifact_path: Relative source path to the desired artifacts file or directory.\n        :param dst_path: Absolute path of the local filesystem destination directory to which\n                         to download the specified artifacts. This directory must already\n                         exist. If unspecified, the artifacts will be downloaded to a new,\n                         uniquely-named\n                         directory on the local filesystem.\n\n        :return: Absolute path of the local filesystem location containing the downloaded\n        artifacts - file/directory.\n        \"\"\"\n\n        hdfs_base_path = _resolve_base_path(self.path, artifact_path)\n        if dst_path and os.path.exists(dst_path):\n            local_dir = os.path.abspath(dst_path)\n        else:\n            local_dir = _tmp_dir(dst_path)\n\n        with hdfs_system(scheme=self.scheme, host=self.host, port=self.port) as hdfs:\n            if not hdfs.isdir(hdfs_base_path):\n                local_path = os.path.join(local_dir, os.path.normpath(artifact_path))\n                _download_hdfs_file(hdfs, hdfs_base_path, local_path)\n                return local_path\n\n            for path, is_dir, _ in self._walk_path(hdfs, hdfs_base_path):\n                relative_path = _relative_path_remote(hdfs_base_path, path)\n                local_path = os.path.join(local_dir, relative_path) if relative_path else local_dir\n\n                if is_dir:\n                    mkdir(local_path)\n                else:\n                    _download_hdfs_file(hdfs, path, local_path)\n            return local_dir\n\n    def _download_file(self, remote_file_path, local_path):\n        raise MlflowException(\"This is not implemented. Should never be called.\")\n\n    def delete_artifacts(self, artifact_path=None):\n        path = posixpath.join(self.path, artifact_path) if artifact_path else self.path\n        with hdfs_system(scheme=self.scheme, host=self.host, port=self.port) as hdfs:\n            hdfs.delete(path, recursive=True)\n\n\n@contextmanager\ndef hdfs_system(scheme, host, port):\n    \"\"\"\n        hdfs system context - Attempt to establish the connection to hdfs\n        and yields HadoopFileSystem\n\n    :param scheme: scheme or use hdfs:// as default\n    :param host: hostname or when relaying on the core-site.xml config use 'default'\n    :param port: port or when relaying on the core-site.xml config use 0\n    \"\"\"\n    import pyarrow as pa\n\n    kerb_ticket = MLFLOW_KERBEROS_TICKET_CACHE.get()\n    kerberos_user = MLFLOW_KERBEROS_USER.get()\n    extra_conf = _parse_extra_conf(MLFLOW_PYARROW_EXTRA_CONF.get())\n\n    if host:\n        host = scheme + \"://\" + host\n    else:\n        host = \"default\"\n\n    connected = pa.hdfs.connect(\n        host=host,\n        port=port or 0,\n        user=kerberos_user,\n        kerb_ticket=kerb_ticket,\n        extra_conf=extra_conf,\n    )\n    yield connected\n    connected.close()\n\n\ndef _resolve_connection_params(artifact_uri):\n    parsed = urllib.parse.urlparse(artifact_uri)\n\n    return parsed.scheme, parsed.hostname, parsed.port, parsed.path\n\n\ndef _resolve_base_path(path, artifact_path):\n    if path == artifact_path:\n        return path\n    if artifact_path:\n        return posixpath.join(path, artifact_path)\n    return path\n\n\ndef _relative_path(base_dir, subdir_path, path_module):\n    relative_path = path_module.relpath(subdir_path, base_dir)\n    return relative_path if relative_path != \".\" else None\n\n\ndef _relative_path_local(base_dir, subdir_path):\n    rel_path = _relative_path(base_dir, subdir_path, os.path)\n    return relative_path_to_artifact_path(rel_path) if rel_path is not None else None\n\n\ndef _relative_path_remote(base_dir, subdir_path):\n    return _relative_path(base_dir, subdir_path, posixpath)\n\n\ndef _tmp_dir(local_path):\n    return os.path.abspath(tempfile.mkdtemp(dir=local_path))\n\n\ndef _download_hdfs_file(hdfs, remote_file_path, local_file_path):\n    # Ensure all required directories exist. Without doing this nested files can't be downloaded.\n    dirs = os.path.dirname(local_file_path)\n    if not os.path.exists(dirs):\n        os.makedirs(dirs)\n    with open(local_file_path, \"wb\") as f:\n        f.write(hdfs.open(remote_file_path, \"rb\").read())\n\n\ndef _parse_extra_conf(extra_conf):\n    if extra_conf:\n\n        def as_pair(config):\n            key, val = config.split(\"=\")\n            return key, val\n\n        list_of_key_val = [as_pair(conf) for conf in extra_conf.split(\",\")]\n        return dict(list_of_key_val)\n    return None\n", "path": "mlflow/store/artifact/hdfs_artifact_repo.py"}], "after_files": [{"content": "import os\nimport posixpath\nimport tempfile\nimport urllib.parse\nfrom contextlib import contextmanager\n\nimport packaging.version\n\nfrom mlflow.entities import FileInfo\nfrom mlflow.environment_variables import (\n    MLFLOW_KERBEROS_TICKET_CACHE,\n    MLFLOW_KERBEROS_USER,\n    MLFLOW_PYARROW_EXTRA_CONF,\n)\nfrom mlflow.exceptions import MlflowException\nfrom mlflow.store.artifact.artifact_repo import ArtifactRepository\nfrom mlflow.utils.file_utils import mkdir, relative_path_to_artifact_path\n\n\nclass HdfsArtifactRepository(ArtifactRepository):\n    \"\"\"\n    Stores artifacts on HDFS.\n\n    This repository is used with URIs of the form ``hdfs:/<path>``. The repository can only be used\n    together with the RestStore.\n    \"\"\"\n\n    def __init__(self, artifact_uri):\n        self.scheme, self.host, self.port, self.path = _resolve_connection_params(artifact_uri)\n        super().__init__(artifact_uri)\n\n    def log_artifact(self, local_file, artifact_path=None):\n        \"\"\"\n            Log artifact in hdfs.\n        :param local_file: source file path\n        :param artifact_path: when specified will attempt to write under artifact_uri/artifact_path\n        \"\"\"\n        hdfs_base_path = _resolve_base_path(self.path, artifact_path)\n\n        with hdfs_system(scheme=self.scheme, host=self.host, port=self.port) as hdfs:\n            _, file_name = os.path.split(local_file)\n            destination = posixpath.join(hdfs_base_path, file_name)\n            with open(local_file, \"rb\") as f:\n                hdfs.upload(destination, f)\n\n    def log_artifacts(self, local_dir, artifact_path=None):\n        \"\"\"\n            Log artifacts in hdfs.\n            Missing remote sub-directories will be created if needed.\n        :param local_dir: source dir path\n        :param artifact_path: when specified will attempt to write under artifact_uri/artifact_path\n        \"\"\"\n        hdfs_base_path = _resolve_base_path(self.path, artifact_path)\n\n        with hdfs_system(scheme=self.scheme, host=self.host, port=self.port) as hdfs:\n            if not hdfs.exists(hdfs_base_path):\n                hdfs.mkdir(hdfs_base_path)\n\n            for subdir_path, _, files in os.walk(local_dir):\n                relative_path = _relative_path_local(local_dir, subdir_path)\n\n                hdfs_subdir_path = (\n                    posixpath.join(hdfs_base_path, relative_path)\n                    if relative_path\n                    else hdfs_base_path\n                )\n\n                if not hdfs.exists(hdfs_subdir_path):\n                    hdfs.mkdir(hdfs_subdir_path)\n\n                for each_file in files:\n                    source = os.path.join(subdir_path, each_file)\n                    destination = posixpath.join(hdfs_subdir_path, each_file)\n                    with open(source, \"rb\") as f:\n                        hdfs.upload(destination, f)\n\n    def list_artifacts(self, path=None):\n        \"\"\"\n        Lists files and directories under artifacts directory for the current run_id.\n        (self.path contains the base path - hdfs:/some/path/run_id/artifacts)\n\n        :param path: Relative source path. Possible subdirectory existing under\n                     hdfs:/some/path/run_id/artifacts\n        :return: List of FileInfos under given path\n        \"\"\"\n        hdfs_base_path = _resolve_base_path(self.path, path)\n\n        with hdfs_system(scheme=self.scheme, host=self.host, port=self.port) as hdfs:\n            paths = []\n            if hdfs.exists(hdfs_base_path):\n                for file_detail in hdfs.ls(hdfs_base_path, detail=True):\n                    file_name = file_detail.get(\"name\")\n\n                    # file_name is hdfs_base_path and not a child of that path\n                    if file_name == hdfs_base_path:\n                        continue\n\n                    # Strip off anything that comes before the artifact root e.g. hdfs://name\n                    offset = file_name.index(self.path)\n                    rel_path = _relative_path_remote(self.path, file_name[offset:])\n                    is_dir = file_detail.get(\"kind\") == \"directory\"\n                    size = file_detail.get(\"size\")\n                    paths.append(FileInfo(rel_path, is_dir=is_dir, file_size=size))\n            return sorted(paths, key=lambda f: paths)\n\n    def _walk_path(self, hdfs, hdfs_path):\n        if hdfs.exists(hdfs_path):\n            if hdfs.isdir(hdfs_path):\n                for subdir, _, files in hdfs.walk(hdfs_path):\n                    if subdir != hdfs_path:\n                        yield subdir, hdfs.isdir(subdir), hdfs.info(subdir).get(\"size\")\n                    for f in files:\n                        file_path = posixpath.join(subdir, f)\n                        yield file_path, hdfs.isdir(file_path), hdfs.info(file_path).get(\"size\")\n            else:\n                yield hdfs_path, False, hdfs.info(hdfs_path).get(\"size\")\n\n    def download_artifacts(self, artifact_path, dst_path=None):\n        \"\"\"\n        Download an artifact file or directory to a local directory/file if applicable, and\n        return a local path for it.\n        The caller is responsible for managing the lifecycle of the downloaded artifacts.\n\n        (self.path contains the base path - hdfs:/some/path/run_id/artifacts)\n\n        :param artifact_path: Relative source path to the desired artifacts file or directory.\n        :param dst_path: Absolute path of the local filesystem destination directory to which\n                         to download the specified artifacts. This directory must already\n                         exist. If unspecified, the artifacts will be downloaded to a new,\n                         uniquely-named\n                         directory on the local filesystem.\n\n        :return: Absolute path of the local filesystem location containing the downloaded\n        artifacts - file/directory.\n        \"\"\"\n\n        hdfs_base_path = _resolve_base_path(self.path, artifact_path)\n        if dst_path and os.path.exists(dst_path):\n            local_dir = os.path.abspath(dst_path)\n        else:\n            local_dir = _tmp_dir(dst_path)\n\n        with hdfs_system(scheme=self.scheme, host=self.host, port=self.port) as hdfs:\n            if not hdfs.isdir(hdfs_base_path):\n                local_path = os.path.join(local_dir, os.path.normpath(artifact_path))\n                _download_hdfs_file(hdfs, hdfs_base_path, local_path)\n                return local_path\n\n            for path, is_dir, _ in self._walk_path(hdfs, hdfs_base_path):\n                relative_path = _relative_path_remote(hdfs_base_path, path)\n                local_path = os.path.join(local_dir, relative_path) if relative_path else local_dir\n\n                if is_dir:\n                    mkdir(local_path)\n                else:\n                    _download_hdfs_file(hdfs, path, local_path)\n            return local_dir\n\n    def _download_file(self, remote_file_path, local_path):\n        raise MlflowException(\"This is not implemented. Should never be called.\")\n\n    def delete_artifacts(self, artifact_path=None):\n        path = posixpath.join(self.path, artifact_path) if artifact_path else self.path\n        with hdfs_system(scheme=self.scheme, host=self.host, port=self.port) as hdfs:\n            hdfs.delete(path, recursive=True)\n\n\n@contextmanager\ndef hdfs_system(scheme, host, port):\n    \"\"\"\n        hdfs system context - Attempt to establish the connection to hdfs\n        and yields HadoopFileSystem\n\n    :param scheme: scheme or use hdfs:// as default\n    :param host: hostname or when relaying on the core-site.xml config use 'default'\n    :param port: port or when relaying on the core-site.xml config use 0\n    \"\"\"\n    import pyarrow\n\n    kerb_ticket = MLFLOW_KERBEROS_TICKET_CACHE.get()\n    kerberos_user = MLFLOW_KERBEROS_USER.get()\n    extra_conf = _parse_extra_conf(MLFLOW_PYARROW_EXTRA_CONF.get())\n\n    if host:\n        host = scheme + \"://\" + host\n    else:\n        host = \"default\"\n    if packaging.version.parse(pyarrow.__version__) < packaging.version.parse(\"2.0.0\"):\n        connected = pyarrow.fs.HadoopFileSystem(\n            host=host,\n            port=port or 0,\n            user=kerberos_user,\n            kerb_ticket=kerb_ticket,\n            extra_conf=extra_conf,\n        )\n    else:\n        connected = pyarrow.hdfs.connect(\n            host=host,\n            port=port or 0,\n            user=kerberos_user,\n            kerb_ticket=kerb_ticket,\n            extra_conf=extra_conf,\n        )\n    yield connected\n    connected.close()\n\n\ndef _resolve_connection_params(artifact_uri):\n    parsed = urllib.parse.urlparse(artifact_uri)\n\n    return parsed.scheme, parsed.hostname, parsed.port, parsed.path\n\n\ndef _resolve_base_path(path, artifact_path):\n    if path == artifact_path:\n        return path\n    if artifact_path:\n        return posixpath.join(path, artifact_path)\n    return path\n\n\ndef _relative_path(base_dir, subdir_path, path_module):\n    relative_path = path_module.relpath(subdir_path, base_dir)\n    return relative_path if relative_path != \".\" else None\n\n\ndef _relative_path_local(base_dir, subdir_path):\n    rel_path = _relative_path(base_dir, subdir_path, os.path)\n    return relative_path_to_artifact_path(rel_path) if rel_path is not None else None\n\n\ndef _relative_path_remote(base_dir, subdir_path):\n    return _relative_path(base_dir, subdir_path, posixpath)\n\n\ndef _tmp_dir(local_path):\n    return os.path.abspath(tempfile.mkdtemp(dir=local_path))\n\n\ndef _download_hdfs_file(hdfs, remote_file_path, local_file_path):\n    # Ensure all required directories exist. Without doing this nested files can't be downloaded.\n    dirs = os.path.dirname(local_file_path)\n    if not os.path.exists(dirs):\n        os.makedirs(dirs)\n    with open(local_file_path, \"wb\") as f:\n        f.write(hdfs.open(remote_file_path, \"rb\").read())\n\n\ndef _parse_extra_conf(extra_conf):\n    if extra_conf:\n\n        def as_pair(config):\n            key, val = config.split(\"=\")\n            return key, val\n\n        list_of_key_val = [as_pair(conf) for conf in extra_conf.split(\",\")]\n        return dict(list_of_key_val)\n    return None\n", "path": "mlflow/store/artifact/hdfs_artifact_repo.py"}]}