fs.hdfs: skip port if not available

dvc/fs/hdfs.py

@@ -27,9 +27,9 @@ class HDFSFileSystem(CallbackMixin, FSSpecWrapper):
 
     def unstrip_protocol(self, path: str) -> str:
         host = self.fs_args["host"]
-        port = self.fs_args["port"]
-        path = path.lstrip("/")
-        return f"hdfs://{host}:{port}/{path}"
+        port = self.fs_args.get("port")
+        netloc = host + (f":{port}" if port else "")
+        return "hdfs://" + netloc + "/" + path.lstrip("/")
 
     @staticmethod
     def _get_kwargs_from_urls(urlpath):

tests/unit/fs/test_hdfs.py

@@ -0,0 +1,16 @@
+import pytest
+
+from dvc.fs import HDFSFileSystem
+
+
+@pytest.mark.parametrize(
+    "fs_args, expected_url",
+    [
+        ({"host": "example.com"}, "hdfs://example.com"),
+        ({"host": "example.com", "port": None}, "hdfs://example.com"),
+        ({"host": "example.com", "port": 8020}, "hdfs://example.com:8020"),
+    ],
+)
+def test_hdfs_unstrip_protocol(fs_args, expected_url):
+    fs = HDFSFileSystem(**fs_args)
+    assert fs.unstrip_protocol("/path") == expected_url + "/path"