[#3379] feat(catalog-hadoop): Add S3 support for Fileset Hadoop catalog

LICENSE.bin

@@ -284,6 +284,7 @@
    Apache Hadoop
    Apache Hadoop Aliyun connector
    Apache Hadoop GCS connector
+   Apache Hadoop AWS connector
    Apache Hadoop Annotatations
    Apache Hadoop Auth
    Apache Hadoop Client Aggregator

build.gradle.kts

@@ -745,7 +745,9 @@ tasks {
       if (!it.name.startsWith("catalog") &&
         !it.name.startsWith("authorization") &&
         !it.name.startsWith("client") && !it.name.startsWith("filesystem") && !it.name.startsWith("spark") && !it.name.startsWith("iceberg") && it.name != "trino-connector" &&
-        it.name != "integration-test" && it.name != "hive-metastore-common" && !it.name.startsWith("flink") && it.name != "gcp-bundle" && it.name != "aliyun-bundle"
+        it.name != "integration-test" && it.name != "bundled-catalog" && !it.name.startsWith("flink") &&
+        it.name != "integration-test" && it.name != "hive-metastore-common" && !it.name.startsWith("flink") &&
+        it.name != "gcp-bundle" && it.name != "aliyun-bundle" && it.name != "aws-bundle"
       ) {
         from(it.configurations.runtimeClasspath)
         into("distribution/package/libs")
@@ -764,8 +766,9 @@ tasks {
         !it.name.startsWith("integration-test") &&
         !it.name.startsWith("flink") &&
         !it.name.startsWith("trino-connector") &&
+        it.name != "bundled-catalog" &&
         it.name != "hive-metastore-common" && it.name != "gcp-bundle" &&
-        it.name != "aliyun-bundle"
+        it.name != "aliyun-bundle" && it.name != "aws-bundle"
       ) {
         dependsOn("${it.name}:build")
         from("${it.name}/build/libs")

bundles/aws-bundle/build.gradle.kts

@@ -0,0 +1,46 @@
+/*
+ * 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.
+ */
+import com.github.jengelman.gradle.plugins.shadow.tasks.ShadowJar
+
+plugins {
+  `maven-publish`
+  id("java")
+  alias(libs.plugins.shadow)
+}
+
+dependencies {
+  compileOnly(project(":catalogs:catalog-hadoop"))
+  compileOnly(libs.hadoop3.common)
+  implementation(libs.hadoop3.aws)
+}
+
+tasks.withType(ShadowJar::class.java) {
+  isZip64 = true
+  configurations = listOf(project.configurations.runtimeClasspath.get())
+  archiveClassifier.set("")
+}
+
+tasks.jar {
+  dependsOn(tasks.named("shadowJar"))
+  archiveClassifier.set("empty")
+}
+
+tasks.compileJava {
+  dependsOn(":catalogs:catalog-hadoop:runtimeJars")
+}

bundles/aws-bundle/src/main/java/org/apache/gravitino/s3/fs/S3FileSystemProvider.java

@@ -0,0 +1,51 @@
+/*
+ * 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.gravitino.s3.fs;
+
+import java.io.IOException;
+import java.util.Map;
+import org.apache.gravitino.catalog.hadoop.fs.FileSystemProvider;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.s3a.S3AFileSystem;
+
+public class S3FileSystemProvider implements FileSystemProvider {
+  @Override
+  public FileSystem getFileSystem(Path path, Map<String, String> config) throws IOException {
+    Configuration configuration = new Configuration();
+    config.forEach(
+        (k, v) -> {
+          configuration.set(k.replace("gravitino.bypass.", ""), v);
+        });
+
+    return S3AFileSystem.newInstance(path.toUri(), configuration);
+  }
+
+  @Override
+  public String scheme() {
+    return "s3a";
+  }
+
+  @Override
+  public String name() {
+    return "s3";
+  }
+}

bundles/aws-bundle/src/main/resources/META-INF/services/org.apache.gravitino.catalog.hadoop.fs.FileSystemProvider

@@ -0,0 +1,20 @@
+#
+# 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.
+#
+
+org.apache.gravitino.s3.fs.S3FileSystemProvider

catalogs/catalog-hadoop/build.gradle.kts

@@ -73,6 +73,7 @@ dependencies {
   testImplementation(project(":integration-test-common", "testArtifacts"))
   testImplementation(project(":server"))
   testImplementation(project(":server-common"))
+  testImplementation(project(":bundles:aws-bundle"))
   testImplementation(project(":bundles:gcp-bundle"))
   testImplementation(project(":bundles:aliyun-bundle"))
 
@@ -161,6 +162,11 @@ tasks.test {
   } else {
     dependsOn(tasks.jar)
   }
+
+  // this task depends on :bundles:aws-bundle:jar
+  dependsOn(":bundles:aws-bundle:jar")
+  dependsOn(":bundles:aliyun-bundle:jar")
+  dependsOn(":bundles:gcp-bundle:jar")
 }
 
 tasks.getByName("generateMetadataFileForMavenJavaPublication") {

catalogs/catalog-hadoop/src/test/java/org/apache/gravitino/catalog/hadoop/integration/test/HadoopCatalogIT.java

@@ -61,19 +61,24 @@ public class HadoopCatalogIT extends BaseIT {
   private static final Logger LOG = LoggerFactory.getLogger(HadoopCatalogIT.class);
   protected static final ContainerSuite containerSuite = ContainerSuite.getInstance();
 
-  public String metalakeName = GravitinoITUtils.genRandomName("CatalogFilesetIT_metalake");
-  public String catalogName = GravitinoITUtils.genRandomName("CatalogFilesetIT_catalog");
-  public static final String SCHEMA_PREFIX = "CatalogFilesetIT_schema";
-  public String schemaName = GravitinoITUtils.genRandomName(SCHEMA_PREFIX);
+  protected String metalakeName = GravitinoITUtils.genRandomName("CatalogFilesetIT_metalake");
+  protected String catalogName = GravitinoITUtils.genRandomName("CatalogFilesetIT_catalog");
+  public final String SCHEMA_PREFIX = "CatalogFilesetIT_schema";
+  protected String schemaName = GravitinoITUtils.genRandomName(SCHEMA_PREFIX);
   protected static final String provider = "hadoop";
-  protected static GravitinoMetalake metalake;
-  protected static Catalog catalog;
-  protected static FileSystem fileSystem;
-  protected static String defaultBaseLocation;
+  protected GravitinoMetalake metalake;
+  protected Catalog catalog;
+  protected FileSystem fileSystem;
+  protected String defaultBaseLocation;
+
+  protected void startNecessaryContainer() {
+    containerSuite.startHiveContainer();
+  }
 
   @BeforeAll
   public void setup() throws IOException {
-    containerSuite.startHiveContainer();
+    startNecessaryContainer();
+
     Configuration conf = new Configuration();
     conf.set("fs.defaultFS", defaultBaseLocation());
     fileSystem = FileSystem.get(conf);

catalogs/catalog-hadoop/src/test/java/org/apache/gravitino/catalog/hadoop/integration/test/HadoopS3CatalogIT.java

@@ -0,0 +1,189 @@
+/*
+ * 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.gravitino.catalog.hadoop.integration.test;
+
+import static org.apache.gravitino.catalog.hadoop.HadoopCatalogPropertiesMetadata.FILESYSTEM_PROVIDERS;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Maps;
+import java.io.IOException;
+import java.net.URI;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.integration.test.container.GravitinoLocalStackContainer;
+import org.apache.gravitino.integration.test.util.GravitinoITUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.shaded.org.awaitility.Awaitility;
+
+@Tag("gravitino-docker-test")
+public class HadoopS3CatalogIT extends HadoopCatalogIT {
+  private static final Logger LOG = LoggerFactory.getLogger(HadoopOSSCatalogIT.class);
+  private String bucketName = "s3-bucket-" + UUID.randomUUID().toString().replace("-", "");
+  private String accessKey;
+  private String secretKey;
+  private String s3Endpoint;
+
+  private GravitinoLocalStackContainer gravitinoLocalStackContainer;
+
+  @VisibleForTesting
+  public void startIntegrationTest() throws Exception {}
+
+  @Override
+  protected void startNecessaryContainer() {
+
+    containerSuite.startLocalStackContainer();
+    gravitinoLocalStackContainer = containerSuite.getLocalStackContainer();
+
+    Awaitility.await()
+        .atMost(60, TimeUnit.SECONDS)
+        .pollInterval(1, TimeUnit.SECONDS)
+        .until(
+            () -> {
+              try {
+                Container.ExecResult result =
+                    gravitinoLocalStackContainer.executeInContainer(
+                        "awslocal", "iam", "create-user", "--user-name", "anonymous");
+                return result.getExitCode() == 0;
+              } catch (Exception e) {
+                LOG.info("LocalStack is not ready yet for: ", e);
+                return false;
+              }
+            });
+
+    gravitinoLocalStackContainer.executeInContainer("awslocal", "s3", "mb", "s3://" + bucketName);
+
+    Container.ExecResult result =
+        gravitinoLocalStackContainer.executeInContainer(
+            "awslocal", "iam", "create-access-key", "--user-name", "anonymous");
+
+    gravitinoLocalStackContainer.executeInContainer(
+        "awslocal",
+        "s3api",
+        "put-bucket-acl",
+        "--bucket",
+        "my-test-bucket",
+        "--acl",
+        "public-read-write");
+
+    // Get access key and secret key from result
+    String[] lines = result.getStdout().split("\n");
+    accessKey = lines[3].split(":")[1].trim().substring(1, 21);
+    secretKey = lines[5].split(":")[1].trim().substring(1, 41);
+
+    LOG.info("Access key: " + accessKey);
+    LOG.info("Secret key: " + secretKey);
+
+    s3Endpoint =
+        String.format("http://%s:%d", gravitinoLocalStackContainer.getContainerIpAddress(), 4566);
+  }
+
+  @BeforeAll
+  public void setup() throws IOException {
+    copyBundleJarsToHadoop("aws-bundle");
+
+    try {
+      super.startIntegrationTest();
+    } catch (Exception e) {
+      throw new RuntimeException("Failed to start integration test", e);
+    }
+
+    startNecessaryContainer();
+
+    metalakeName = GravitinoITUtils.genRandomName("CatalogFilesetIT_metalake");
+    catalogName = GravitinoITUtils.genRandomName("CatalogFilesetIT_catalog");
+    schemaName = GravitinoITUtils.genRandomName("CatalogFilesetIT_schema");
+
+    schemaName = GravitinoITUtils.genRandomName(SCHEMA_PREFIX);
+    Configuration conf = new Configuration();
+
+    conf.set("fs.s3a.access.key", accessKey);
+    conf.set("fs.s3a.secret.key", secretKey);
+    conf.set("fs.s3a.endpoint", s3Endpoint);
+    conf.set(
+        "fs.s3a.aws.credentials.provider", "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider");
+    fileSystem = FileSystem.get(URI.create(String.format("s3a://%s", bucketName)), conf);
+
+    createMetalake();
+    createCatalog();
+    createSchema();
+  }
+
+  @AfterAll
+  public void stop() throws IOException {
+    Catalog catalog = metalake.loadCatalog(catalogName);
+    catalog.asSchemas().dropSchema(schemaName, true);
+    metalake.dropCatalog(catalogName, true);
+    client.dropMetalake(metalakeName);
+
+    try {
+      closer.close();
+    } catch (Exception e) {
+      LOG.error("Failed to close CloseableGroup", e);
+    }
+  }
+
+  protected String defaultBaseLocation() {
+    if (defaultBaseLocation == null) {
+      try {
+        Path bucket =
+            new Path(
+                String.format(
+                    "s3a://%s/%s", bucketName, GravitinoITUtils.genRandomName("CatalogFilesetIT")));
+        if (!fileSystem.exists(bucket)) {
+          fileSystem.mkdirs(bucket);
+        }
+
+        defaultBaseLocation = bucket.toString();
+      } catch (IOException e) {
+        throw new RuntimeException("Failed to create default base location", e);
+      }
+    }
+
+    return defaultBaseLocation;
+  }
+
+  protected void createCatalog() {
+    Map<String, String> map = Maps.newHashMap();
+    map.put("gravitino.bypass.fs.s3a.access.key", accessKey);
+    map.put("gravitino.bypass.fs.s3a.secret.key", secretKey);
+    map.put("gravitino.bypass.fs.s3a.endpoint", s3Endpoint);
+    map.put(
+        "gravitino.bypass.fs.s3a.aws.credentials.provider",
+        "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider");
+    map.put(FILESYSTEM_PROVIDERS, "s3");
+
+    metalake.createCatalog(catalogName, Catalog.Type.FILESET, provider, "comment", map);
+
+    catalog = metalake.loadCatalog(catalogName);
+  }
+
+  protected String generateLocation(String filesetName) {
+    return String.format("%s/%s", defaultBaseLocation, filesetName);
+  }
+}