HADOOP-16939 fs.s3a.authoritative.path should support multiple FS URIs (#1914)

add unit test, new ITest and then fix the issue: different schema, bucket == skip

factored out the underlying logic for unit testing; also moved
maybeAddTrailingSlash to S3AUtils (while retaining/forwarnding existing method
in S3AFS).

tested: london, sole failure is
testListingDelete[auth=true](org.apache.hadoop.fs.s3a.ITestS3GuardOutOfBandOperations)

filed HADOOP-16853

Change-Id: I4b8d0024469551eda0ec70b4968cba4abed405ed

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

@@ -974,11 +974,7 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
    */
   @InterfaceAudience.Private
   public String maybeAddTrailingSlash(String key) {
-    if (!key.isEmpty() && !key.endsWith("/")) {
-      return key + '/';
-    } else {
-      return key;
-    }
+    return S3AUtils.maybeAddTrailingSlash(key);
   }
 
   /**

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

@@ -1726,6 +1726,21 @@ public final class S3AUtils {
     return conf.get(FS_S3A_BUCKET_PREFIX + bucket + '.' + baseKey);
   }
 
+  /**
+   * Turns a path (relative or otherwise) into an S3 key, adding a trailing
+   * "/" if the path is not the root <i>and</i> does not already have a "/"
+   * at the end.
+   *
+   * @param key s3 key or ""
+   * @return the with a trailing "/", or, if it is the root key, "",
+   */
+  public static String maybeAddTrailingSlash(String key) {
+    if (!key.isEmpty() && !key.endsWith("/")) {
+      return key + '/';
+    } else {
+      return key;
+    }
+  }
 
   /**
    * Path filter which ignores any file which starts with . or _.

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

@@ -30,6 +30,7 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import javax.annotation.Nullable;
@@ -974,13 +975,43 @@ public final class S3Guard {
   }
 
   public static Collection<String> getAuthoritativePaths(S3AFileSystem fs) {
+    return getAuthoritativePaths(
+        fs.getUri(),
+        fs.getConf(),
+        p -> fs.maybeAddTrailingSlash(fs.qualify(p).toString()));
+  }
+
+  /**
+   * Get the authoritative paths of a filesystem.
+   *
+   * @param uri FS URI
+   * @param conf configuration
+   * @param qualifyToDir a qualification operation
+   * @return list of URIs valid for this FS.
+   */
+  @VisibleForTesting
+  static Collection<String> getAuthoritativePaths(
+      final URI uri,
+      final Configuration conf,
+      final Function<Path, String> qualifyToDir) {
     String[] rawAuthoritativePaths =
-        fs.getConf().getTrimmedStrings(AUTHORITATIVE_PATH, DEFAULT_AUTHORITATIVE_PATH);
+        conf.getTrimmedStrings(AUTHORITATIVE_PATH, DEFAULT_AUTHORITATIVE_PATH);
     Collection<String> authoritativePaths = new ArrayList<>();
     if (rawAuthoritativePaths.length > 0) {
       for (int i = 0; i < rawAuthoritativePaths.length; i++) {
-        Path qualified = fs.qualify(new Path(rawAuthoritativePaths[i]));
-        authoritativePaths.add(fs.maybeAddTrailingSlash(qualified.toString()));
+        Path path = new Path(rawAuthoritativePaths[i]);
+        URI pathURI = path.toUri();
+        if (pathURI.getAuthority() != null &&
+            !pathURI.getAuthority().equals(uri.getAuthority())) {
+          // skip on auth
+          continue;
+        }
+        if (pathURI.getScheme() != null &&
+            !pathURI.getScheme().equals(uri.getScheme())) {
+          // skip on auth
+          continue;
+        }
+        authoritativePaths.add(qualifyToDir.apply(path));
       }
     }
     return authoritativePaths;

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

@@ -233,6 +233,28 @@ public class ITestAuthoritativePath extends AbstractS3ATestBase {
     }
   }
 
+  @Test
+  public void testAuthPathWithOtherBucket() throws Exception {
+    Path authPath;
+    Path nonAuthPath;
+    S3AFileSystem fs = null;
+    String landsat = "s3a://landsat-pds/data";
+    String decoy2 = "/decoy2";
+
+    try {
+      authPath = new Path(testRoot, "testMultiAuthPath-first");
+      nonAuthPath = new Path(testRoot, "nonAuth-1");
+      fs = createMultiPathAuthFS(authPath.toString(), landsat, decoy2);
+      assertTrue("No S3Guard store for partially authoritative FS",
+          fs.hasMetadataStore());
+
+      runTestInsidePath(fs, authPath);
+      runTestOutsidePath(fs, nonAuthPath);
+    } finally {
+      cleanUpFS(fs);
+    }
+  }
+
   @Test
   public void testMultiAuthPath() throws Exception {
     Path authPath;

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

@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.s3a.s3guard;
+
+import java.net.URI;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.s3a.S3AUtils;
+import org.apache.hadoop.test.AbstractHadoopTestBase;
+
+import static org.apache.hadoop.fs.s3a.Constants.AUTHORITATIVE_PATH;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Unit tests of auth path resolution.
+ */
+public class TestAuthoritativePath extends AbstractHadoopTestBase {
+
+  private final Path root = new Path("/");
+
+  private URI fsUri;
+
+  private static final String BASE = "s3a://bucket";
+
+  @Before
+  public void setup() throws Exception {
+    fsUri = new URI(BASE +"/");
+  }
+
+  private Configuration authPathsConf(String... paths) {
+    Configuration conf = new Configuration(false);
+    conf.set(AUTHORITATIVE_PATH, String.join(",", paths));
+    return conf;
+  }
+
+  @Test
+  public void testResolution() throws Throwable {
+    assertAuthPaths(l("/one"), "/one/");
+  }
+
+  @Test
+  public void testResolutionWithFQP() throws Throwable {
+    assertAuthPaths(l("/one/",
+        BASE + "/two/"),
+        "/one/", "/two/");
+  }
+  @Test
+  public void testOtherBucket() throws Throwable {
+    assertAuthPaths(l("/one/",
+        "s3a://landsat-pds/",
+        BASE + "/two/"),
+        "/one/", "/two/");
+  }
+
+  @Test
+  public void testOtherScheme() throws Throwable {
+    assertAuthPaths(l("/one/",
+        "s3a://landsat-pds/",
+        "http://bucket/two/"),
+        "/one/");
+  }
+
+  /**
+   * Get the auth paths; qualification is through
+   * Path.makeQualified not the FS near-equivalent.
+   * @param conf configuration
+   * @return list of auth paths.
+   */
+  private Collection<String> getAuthoritativePaths(
+      Configuration conf) {
+
+    return S3Guard.getAuthoritativePaths(fsUri, conf,
+        p -> {
+          Path q = p.makeQualified(fsUri, root);
+          assertThat(q.toUri().getAuthority())
+              .describedAs("Path %s", q)
+              .isEqualTo(fsUri.getAuthority());
+          return S3AUtils.maybeAddTrailingSlash(q.toString());
+        });
+  }
+
+  /**
+   * take a varargs list and and return as an array.
+   * @param s source
+   * @return the values
+   */
+  private String[] l(String...s) {
+    return s;
+  }
+
+  /**
+   * Assert that the authoritative paths from a source list
+   * are that expected.
+   * @param src source entries to set as auth paths
+   * @param expected the list of auth paths for a filesystem
+   */
+  private void assertAuthPaths(String[] src, String...expected) {
+    Configuration conf = authPathsConf(src);
+    List<String> collect = Arrays.stream(expected)
+        .map(s -> BASE + s)
+        .collect(Collectors.toList());
+    Collection<String> paths = getAuthoritativePaths(conf);
+    assertThat(paths)
+        .containsExactlyInAnyOrderElementsOf(collect);
+  }
+
+}