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

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

@@ -267,9 +267,10 @@ 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();
+    final URI endpoint = getS3Endpoint(endpointStr, conf);
 
-    String configuredRegion = parameters.getRegion();
+    final String configuredRegion = parameters.getRegion();
     Region region = null;
     String origin = "";
 
@@ -289,17 +290,36 @@ private <BuilderT extends S3BaseClientBuilder<BuilderT, ClientT>, ClientT> void
     builder.fipsEnabled(fipsEnabled);
 
     if (endpoint != null) {
+      boolean overrideEndpoint = true;
       checkArgument(!fipsEnabled,
           "%s : %s", ERROR_ENDPOINT_WITH_FIPS, endpoint);
-      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);
+      // No region was configured or the endpoint is central,
+      // determine the region from the endpoint.
+      if (region == null || endpointEndsWithCentral) {
+        region = getS3RegionFromEndpoint(endpointStr,
+            endpointEndsWithCentral);
         if (region != null) {
           origin = "endpoint";
+          if (endpointEndsWithCentral) {
+            // No need to override endpoint with "s3.amazonaws.com".
+            // Let the client take care of endpoint resolution. Overriding
+            // the endpoint with "s3.amazonaws.com" causes 400 Bad Request
+            // errors for non-existent buckets and objects.
+            // ref: https://github.com/aws/aws-sdk-java-v2/issues/4846
+            overrideEndpoint = false;
+            builder.crossRegionAccessEnabled(true);
+            origin = "origin with cross region access";
+            LOG.debug("Enabling cross region access for endpoint {}",
+                endpointStr);
+          }
         }
       }
-      LOG.debug("Setting endpoint to {}", endpoint);
+      if (overrideEndpoint) {
+        builder.endpointOverride(endpoint);
+        LOG.debug("Setting endpoint to {}", endpoint);
+      }
     }
 
     if (region != null) {
@@ -354,20 +374,32 @@ 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(final String endpoint,
+      final 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;
+    // Select default region here to enable cross-region access.
+    // If both "fs.s3a.endpoint" and "fs.s3a.endpoint.region" are empty,
+    // Spark sets "fs.s3a.endpoint" to "s3.amazonaws.com".
+    // This applies to Spark versions with the changes of SPARK-35878.
+    // ref:
+    // https://github.com/apache/spark/blob/v3.5.0/core/
+    // src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala#L528
+    // If we do not allow cross region access, Spark would not be able to
+    // access any bucket that is not present in the given region.
+    // Hence, we should use default region us-east-2 to allow cross-region
+    // access.
+    return Region.of(AWS_S3_DEFAULT_REGION);
   }
 
 }

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

@@ -38,10 +38,14 @@
 import software.amazon.awssdk.services.s3.model.HeadBucketResponse;
 
 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 org.apache.hadoop.fs.s3a.statistics.impl.EmptyS3AStatisticsContext;
 
 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.Constants.PATH_STYLE_ACCESS;
 import static org.apache.hadoop.fs.s3a.DefaultS3ClientFactory.ERROR_ENDPOINT_WITH_FIPS;
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
@@ -146,7 +150,21 @@ 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_EAST_2, false);
+
+    expectInterceptorException(client);
+
+    client = createS3Client(conf, CENTRAL_ENDPOINT, US_EAST_1, US_EAST_2, false);
 
     expectInterceptorException(client);
   }
@@ -257,6 +275,65 @@ public void testWithVPCE() throws Throwable {
     expectInterceptorException(client);
   }
 
+  @Test
+  public void testCentralEndpointCrossRegionAccess() throws Throwable {
+    describe("Create bucket on different region and access it using central endpoint");
+    final Configuration conf = getConfiguration();
+    removeBaseAndBucketOverrides(conf, ENDPOINT);
+
+    final Configuration newConf = new Configuration(conf);
+
+    newConf.set(ENDPOINT, CENTRAL_ENDPOINT);
+
+    newFS = new S3AFileSystem();
+    newFS.initialize(getFileSystem().getUri(), newConf);
+
+    assertOpsUsingNewFs();
+  }
+
+  @Test
+  public void testCentralEndpointWithNullRegionCrossRegionAccess() throws Throwable {
+    describe(
+        "Create bucket on different region and access it using central endpoint and null region");
+    final Configuration conf = getConfiguration();
+    removeBaseAndBucketOverrides(conf, ENDPOINT, AWS_REGION);
+
+    final Configuration newConf = new Configuration(conf);
+
+    newConf.set(ENDPOINT, CENTRAL_ENDPOINT);
+
+    newFS = new S3AFileSystem();
+    newFS.initialize(getFileSystem().getUri(), newConf);
+
+    assertOpsUsingNewFs();
+  }
+
+  private void assertOpsUsingNewFs() throws IOException {
+    final String file = getMethodName();
+    final Path basePath = methodPath();
+    final Path srcDir = new Path(basePath, "srcdir");
+    newFS.mkdirs(srcDir);
+    Path srcFilePath = new Path(srcDir, file);
+
+    try (FSDataOutputStream out = newFS.create(srcFilePath)) {
+      out.write(new byte[] {1, 2, 3});
+    }
+
+    ContractTestUtils.assertIsFile(getFileSystem(), srcFilePath);
+    ContractTestUtils.assertIsFile(newFS, srcFilePath);
+
+    newFS.delete(srcDir, true);
+
+    Assertions
+        .assertThat(newFS.exists(srcFilePath))
+        .describedAs("Existence of file: " + srcFilePath)
+        .isFalse();
+    Assertions
+        .assertThat(getFileSystem().exists(srcFilePath))
+        .describedAs("Existence of file: " + srcFilePath)
+        .isFalse();
+  }
+
   private final class RegionInterceptor implements ExecutionInterceptor {
     private final String endpoint;
     private final String region;
@@ -272,7 +349,7 @@ private final class RegionInterceptor implements ExecutionInterceptor {
     public void beforeExecution(Context.BeforeExecution context,
         ExecutionAttributes executionAttributes)  {
 
-      if (endpoint != null) {
+      if (endpoint != null && !endpoint.endsWith(CENTRAL_ENDPOINT)) {
         Assertions.assertThat(
                 executionAttributes.getAttribute(AwsExecutionAttribute.ENDPOINT_OVERRIDDEN))
             .describedAs("Endpoint not overridden").isTrue();