[SPARK-44845][YARN][DEPLOY] Fix file system uri comparison function

### What changes were proposed in this pull request?
What changes were proposed in this pull request?

### Why are the changes needed?

In the org.apache.spark.deploy.yarn.Client#compareUri method, hdfs://hadoop81:8020 and hdfs://192.168.0.81:8020 are regarded as different file systems (hadoop81 corresponds to 192.168.0.81). The specific reason is that in the last pr, different URIs of user information are also regarded as different file systems. Uri.getauthority is used to determine the user information, but authority contains the host so the URI above must be different from authority. To determine whether the user authentication information is different, you only need to determine URI.getUserInfo.

the last pr and issue link:
https://issues.apache.org/jira/browse/SPARK-22587

apache#19885

### Does this PR introduce _any_ user-facing change?

### How was this patch tested?

Closes apache#42529 from zekai-li/master.

Authored-by: zekai-li <[email protected]>
Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com>

Author: zekai-li

resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala

@@ -1618,9 +1618,10 @@ private[spark] object Client extends Logging {
       return false
     }
 
-    val srcAuthority = srcUri.getAuthority()
-    val dstAuthority = dstUri.getAuthority()
-    if (srcAuthority != null && !srcAuthority.equalsIgnoreCase(dstAuthority)) {
+    val srcUserInfo = Option(srcUri.getUserInfo).getOrElse("")
+    val dstUserInfo = Option(dstUri.getUserInfo).getOrElse("")
+
+    if (!srcUserInfo.equals(dstUserInfo)) {
       return false
     }
 

resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ClientSuite.scala

@@ -675,7 +675,8 @@ class ClientSuite extends SparkFunSuite
     ("files URI match test2", "file:///c:file1", "file://c:file2"),
     ("files URI match test3", "file://host/file1", "file://host/file2"),
     ("wasb URI match test", "wasb://bucket1@user", "wasb://bucket1@user/"),
-    ("hdfs URI match test", "hdfs:/path1", "hdfs:/path1")
+    ("hdfs URI match test1", "hdfs:/path1", "hdfs:/path1"),
+    ("hdfs URI match test2", "hdfs://localhost:8080", "hdfs://127.0.0.1:8080")
   )
 
   matching.foreach { t =>
@@ -691,7 +692,7 @@ class ClientSuite extends SparkFunSuite
     ("files URI unmatch test3", "file://host/file1", "file://host2/file2"),
     ("wasb URI unmatch test1", "wasb://bucket1@user", "wasb://bucket2@user/"),
     ("wasb URI unmatch test2", "wasb://bucket1@user", "wasb://bucket1@user2/"),
-    ("s3 URI unmatch test", "s3a://user@pass:bucket1/", "s3a://user2@pass2:bucket1/"),
+    ("s3 URI unmatch test", "s3a://user:pass@bucket1/", "s3a://user2:pass2@bucket1/"),
     ("hdfs URI unmatch test1", "hdfs://namenode1/path1", "hdfs://namenode1:8080/path2"),
     ("hdfs URI unmatch test2", "hdfs://namenode1:8020/path1", "hdfs://namenode1:8080/path2")
   )