HADOOP-19044. AWS SDK V2 - Update S3A region logic

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

@@ -267,7 +267,8 @@ protected ClientOverrideConfiguration.Builder createClientOverrideConfiguration(
    */
   private <BuilderT extends S3BaseClientBuilder<BuilderT, ClientT>, ClientT> void configureEndpointAndRegion(
       BuilderT builder, S3ClientCreationParameters parameters, Configuration conf) {
-    URI endpoint = getS3Endpoint(parameters.getEndpoint(), conf);
+    final String endpointStr = parameters.getEndpoint();
+    URI endpoint = getS3Endpoint(endpointStr, conf);
 
     String configuredRegion = parameters.getRegion();
     Region region = null;
@@ -294,9 +295,14 @@ private <BuilderT extends S3BaseClientBuilder<BuilderT, ClientT>, ClientT> void
       builder.endpointOverride(endpoint);
       // No region was configured, try to determine it from the endpoint.
       if (region == null) {
-        region = getS3RegionFromEndpoint(parameters.getEndpoint());
+        boolean endpointEndsWithCentral = endpointStr.endsWith(CENTRAL_ENDPOINT);
+        region = getS3RegionFromEndpoint(endpointStr, endpointEndsWithCentral);
         if (region != null) {
           origin = "endpoint";
+          if (endpointEndsWithCentral) {
+            builder.crossRegionAccessEnabled(true);
+            LOG.debug("Enabling cross region access for endpoint {}", endpointStr);
+          }
         }
       }
       LOG.debug("Setting endpoint to {}", endpoint);
@@ -354,20 +360,21 @@ private static URI getS3Endpoint(String endpoint, final Configuration conf) {
 
   /**
    * Parses the endpoint to get the region.
-   * If endpoint is the central one, use US_EAST_1.
+   * If endpoint is the central one, use US_EAST_2.
    *
    * @param endpoint the configure endpoint.
+   * @param endpointEndsWithCentral true if the endpoint is configured as central.
    * @return the S3 region, null if unable to resolve from endpoint.
    */
-  private static Region getS3RegionFromEndpoint(String endpoint) {
+  private static Region getS3RegionFromEndpoint(String endpoint, boolean endpointEndsWithCentral) {
 
-    if(!endpoint.endsWith(CENTRAL_ENDPOINT)) {
+    if (!endpointEndsWithCentral) {
       LOG.debug("Endpoint {} is not the default; parsing", endpoint);
       return AwsHostNameUtils.parseSigningRegion(endpoint, S3_SERVICE_NAME).orElse(null);
     }
 
-    // endpoint is for US_EAST_1;
-    return Region.US_EAST_1;
+    // endpoint is for US_EAST_2;
+    return Region.US_EAST_2;
   }
 
 }

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

@@ -0,0 +1,65 @@
+/*
+ * 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;
+
+import org.junit.Test;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.contract.ContractTestUtils;
+
+import static org.apache.hadoop.fs.s3a.Constants.AWS_REGION;
+import static org.apache.hadoop.fs.s3a.Constants.CENTRAL_ENDPOINT;
+import static org.apache.hadoop.fs.s3a.Constants.ENDPOINT;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
+
+/**
+ * Test to verify cross region bucket access.
+ */
+public class ITestS3ACrossRegionAccess extends AbstractS3ATestBase {
+
+  @Test
+  public void testCentralEndpointCrossRegionAccess() throws Throwable {
+    describe("Create bucket on different region and access it using central endpoint");
+    Configuration conf = getConfiguration();
+    removeBaseAndBucketOverrides(conf, ENDPOINT, AWS_REGION);
+
+    Configuration newConf = new Configuration(conf);
+
+    newConf.set(ENDPOINT, CENTRAL_ENDPOINT);
+
+    try (S3AFileSystem newFs = new S3AFileSystem()) {
+      newFs.initialize(getFileSystem().getUri(), newConf);
+
+      final String file = getMethodName();
+      Path basePath = new Path("basePath-" + getMethodName());
+      final Path srcDir = new Path(basePath, "srcdir");
+      newFs.mkdirs(srcDir);
+      Path src = new Path(srcDir, file);
+
+      try (FSDataOutputStream out = newFs.create(src)) {
+        out.write(new byte[] {1, 2, 3, 4, 5});
+      }
+      ContractTestUtils.assertIsFile(getFileSystem(), new Path(srcDir, file));
+      newFs.delete(srcDir, true);
+    }
+  }
+
+}

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

@@ -146,7 +146,17 @@ public void testCentralEndpoint() throws Throwable {
     describe("Create a client with the central endpoint");
     Configuration conf = getConfiguration();
 
-    S3Client client = createS3Client(conf, CENTRAL_ENDPOINT, null, US_EAST_1, false);
+    S3Client client = createS3Client(conf, CENTRAL_ENDPOINT, null, US_EAST_2, false);
+
+    expectInterceptorException(client);
+  }
+
+  @Test
+  public void testCentralEndpointWithRegion() throws Throwable {
+    describe("Create a client with the central endpoint but also specify region");
+    Configuration conf = getConfiguration();
+
+    S3Client client = createS3Client(conf, CENTRAL_ENDPOINT, US_WEST_2, US_WEST_2, false);
 
     expectInterceptorException(client);
   }