HADOOP-16409. Allow authoritative mode on non-qualified paths.

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

@@ -2421,7 +2421,6 @@ public FileStatus[] innerListStatus(Path f) throws FileNotFoundException,
         result.add(files.next());
       }
       // merge the results. This will update the store as needed
-
       return S3Guard.dirListingUnion(metadataStore, path, result, dirMeta,
           allowAuthoritative, ttlTimeProvider);
     } else {
@@ -3810,7 +3809,6 @@ private RemoteIterator<S3ALocatedFileStatus> innerListFiles(Path f, boolean
           final PathMetadata pm = metadataStore.get(path, true);
           // shouldn't need to check pm.isDeleted() because that will have
           // been caught by getFileStatus above.
-
           MetadataStoreListFilesIterator metadataStoreListFilesIterator =
               new MetadataStoreListFilesIterator(metadataStore, pm,
                   allowAuthoritative);

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/DynamoDBMetadataStore.java

@@ -1321,6 +1321,7 @@ public void put(
       final DirListingMetadata meta,
       @Nullable final BulkOperationState operationState) throws IOException {
     LOG.debug("Saving to table {} in region {}: {}", tableName, region, meta);
+
     // directory path
     Path path = meta.getPath();
     DDBPathMetadata ddbPathMeta =

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/S3Guard.java

@@ -787,13 +787,12 @@ public static Collection<String> getAuthoritativePaths(S3AFileSystem fs) {
 
   public static boolean allowAuthoritative(Path p, S3AFileSystem fs,
       boolean authMetadataStore, Collection<String> authPaths) {
-    String haystack = fs.maybeAddTrailingSlash(p.toString());
+    String haystack = fs.maybeAddTrailingSlash(fs.qualify(p).toString());
     if (authMetadataStore) {
       return true;
     }
     if (!authPaths.isEmpty()) {
       for (String needle : authPaths) {
-
         if (haystack.startsWith(needle)) {
           return true;
         }

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestAuthoritativePath.java

@@ -294,6 +294,10 @@ public void testPrefixVsDirectory() throws Exception {
       Path directoryMatch = new Path(testRoot, "/auth/oritative");
       assertTrue(S3Guard.allowAuthoritative(directoryMatch, fs,
           false, authPaths));
+
+      Path unqualifiedMatch = new Path(testRoot.toUri().getPath(), "/auth/oritative");
+      assertTrue(S3Guard.allowAuthoritative(unqualifiedMatch, fs,
+          false, authPaths));
     } finally {
       cleanUpFS(fs);
     }