Parse S3 VPC endpoints for the region

Author: Shintaro Onuma

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

@@ -21,6 +21,8 @@
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 import org.apache.hadoop.classification.VisibleForTesting;
 import org.apache.hadoop.fs.s3a.impl.AWSClientConfig;
@@ -82,6 +84,9 @@ public class DefaultS3ClientFactory extends Configured
 
   private static final String S3_SERVICE_NAME = "s3";
 
+  private static final Pattern VPC_ENDPOINT_PATTERN =
+          Pattern.compile("^(?:.+\\.)?([a-z0-9-]+)\\.vpce\\.");
+
   /**
    * Subclasses refer to this.
    */
@@ -361,6 +366,13 @@ private static URI getS3Endpoint(String endpoint, final Configuration conf) {
    */
   private static Region getS3RegionFromEndpoint(String endpoint) {
 
+    // S3 VPC endpoint parsing
+    Matcher matcher = VPC_ENDPOINT_PATTERN.matcher(endpoint);
+    if(matcher.find()) {
+      LOG.debug("Endpoint {} is vpc endpoint; parsing", endpoint);
+      return Region.of(matcher.group(1));
+    }
+
     if(!endpoint.endsWith(CENTRAL_ENDPOINT)) {
       LOG.debug("Endpoint {} is not the default; parsing", endpoint);
       return AwsHostNameUtils.parseSigningRegion(endpoint, S3_SERVICE_NAME).orElse(null);

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

@@ -82,6 +82,8 @@ public class ITestS3AEndpointRegion extends AbstractS3ATestBase {
 
   private static final String VPC_ENDPOINT = "vpce-1a2b3c4d-5e6f.s3.us-west-2.vpce.amazonaws.com";
 
+  private static final String CN_VPC_ENDPOINT = "vpce-1a2b3c4d-5e6f.s3.cn-northwest-1.vpce.amazonaws.com.cn";
+
   public static final String EXCEPTION_THROWN_BY_INTERCEPTOR = "Exception thrown by interceptor";
 
   /**
@@ -247,7 +249,6 @@ public void testWithGovCloudEndpoint() throws Throwable {
   }
 
   @Test
-  @Ignore("Pending HADOOP-18938. S3A region logic to handle vpce and non standard endpoints")
   public void testWithVPCE() throws Throwable {
     describe("Test with vpc endpoint");
     Configuration conf = getConfiguration();
@@ -257,6 +258,16 @@ public void testWithVPCE() throws Throwable {
     expectInterceptorException(client);
   }
 
+  @Test
+  public void testWithChinaVPCE() throws Throwable {
+    describe("Test with china vpc endpoint");
+    Configuration conf = getConfiguration();
+
+    S3Client client = createS3Client(conf, CN_VPC_ENDPOINT, null, CN_NORTHWEST_1, false);
+
+    expectInterceptorException(client);
+  }
+
   private final class RegionInterceptor implements ExecutionInterceptor {
     private final String endpoint;
     private final String region;