[#5830] fix(clients): add error handling for no tag in cli.

build.gradle.kts

@@ -779,7 +779,7 @@ tasks {
         !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 != "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" && it.name != "azure-bundle"
+        it.name != "gcp-bundle" && it.name != "aliyun-bundle" && it.name != "aws-bundle" && it.name != "azure-bundle" && it.name != "hadoop-common"
       ) {
         from(it.configurations.runtimeClasspath)
         into("distribution/package/libs")
@@ -802,7 +802,7 @@ tasks {
         it.name != "bundled-catalog" &&
         it.name != "hive-metastore-common" && it.name != "gcp-bundle" &&
         it.name != "aliyun-bundle" && it.name != "aws-bundle" && it.name != "azure-bundle" &&
-        it.name != "docs"
+        it.name != "hadoop-common" && it.name != "docs"
       ) {
         dependsOn("${it.name}:build")
         from("${it.name}/build/libs")

bundles/aliyun-bundle/build.gradle.kts

@@ -29,6 +29,9 @@ dependencies {
   compileOnly(project(":core"))
   compileOnly(project(":catalogs:catalog-common"))
   compileOnly(project(":catalogs:catalog-hadoop"))
+  compileOnly(project(":catalogs:hadoop-common")) {
+    exclude("*")
+  }
   compileOnly(libs.hadoop3.common)
 
   implementation(libs.aliyun.credentials.sdk)

bundles/aws-bundle/build.gradle.kts

@@ -29,6 +29,9 @@ dependencies {
   compileOnly(project(":core"))
   compileOnly(project(":catalogs:catalog-common"))
   compileOnly(project(":catalogs:catalog-hadoop"))
+  compileOnly(project(":catalogs:hadoop-common")) {
+    exclude("*")
+  }
   compileOnly(libs.hadoop3.common)
 
   implementation(libs.aws.iam)

bundles/aws-bundle/src/main/java/org/apache/gravitino/s3/credential/S3TokenProvider.java

@@ -104,6 +104,13 @@ private StsClient createStsClient(S3CredentialConfig s3CredentialConfig) {
     if (StringUtils.isNotBlank(region)) {
       builder.region(Region.of(region));
     }
+    String stsEndpoint = s3CredentialConfig.stsEndpoint();
+    // If the user does not set a value or provides an blank string, we treat as unspecified.
+    // The goal is to pass a custom endpoint to the `builder` only when the user specifies a
+    // non-blank value.
+    if (StringUtils.isNotBlank(stsEndpoint)) {
+      builder.endpointOverride(URI.create(stsEndpoint));
+    }
     return builder.build();
   }
 

bundles/azure-bundle/build.gradle.kts

@@ -28,6 +28,9 @@ dependencies {
   compileOnly(project(":api"))
   compileOnly(project(":core"))
   compileOnly(project(":catalogs:catalog-hadoop"))
+  compileOnly(project(":catalogs:hadoop-common")) {
+    exclude("*")
+  }
 
   compileOnly(libs.hadoop3.common)
 

bundles/gcp-bundle/build.gradle.kts

@@ -29,6 +29,9 @@ dependencies {
   compileOnly(project(":core"))
   compileOnly(project(":catalogs:catalog-common"))
   compileOnly(project(":catalogs:catalog-hadoop"))
+  compileOnly(project(":catalogs:hadoop-common")) {
+    exclude("*")
+  }
 
   compileOnly(libs.hadoop3.common)
 

catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonConstants.java

@@ -0,0 +1,57 @@
+/*
+ * 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.lakehouse.paimon;
+
+public class PaimonConstants {
+
+  // Paimon catalog properties constants
+  public static final String CATALOG_BACKEND = "catalog-backend";
+  public static final String METASTORE = "metastore";
+  public static final String URI = "uri";
+  public static final String WAREHOUSE = "warehouse";
+  public static final String CATALOG_BACKEND_NAME = "catalog-backend-name";
+
+  public static final String GRAVITINO_JDBC_USER = "jdbc-user";
+  public static final String PAIMON_JDBC_USER = "jdbc.user";
+
+  public static final String GRAVITINO_JDBC_PASSWORD = "jdbc-password";
+  public static final String PAIMON_JDBC_PASSWORD = "jdbc.password";
+
+  public static final String GRAVITINO_JDBC_DRIVER = "jdbc-driver";
+
+  // S3 properties needed by Paimon
+  public static final String S3_ENDPOINT = "s3.endpoint";
+  public static final String S3_ACCESS_KEY = "s3.access-key";
+  public static final String S3_SECRET_KEY = "s3.secret-key";
+
+  // OSS related properties
+  public static final String OSS_ENDPOINT = "fs.oss.endpoint";
+  public static final String OSS_ACCESS_KEY = "fs.oss.accessKeyId";
+  public static final String OSS_SECRET_KEY = "fs.oss.accessKeySecret";
+
+  // Iceberg Table properties constants
+  public static final String COMMENT = "comment";
+  public static final String OWNER = "owner";
+  public static final String BUCKET_KEY = "bucket-key";
+  public static final String MERGE_ENGINE = "merge-engine";
+  public static final String SEQUENCE_FIELD = "sequence.field";
+  public static final String ROWKIND_FIELD = "rowkind.field";
+  public static final String PRIMARY_KEY = "primary-key";
+  public static final String PARTITION = "partition";
+}

catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonPropertiesUtils.java

@@ -0,0 +1,95 @@
+/*
+ * 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.lakehouse.paimon;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+import org.apache.gravitino.storage.OSSProperties;
+import org.apache.gravitino.storage.S3Properties;
+
+public class PaimonPropertiesUtils {
+
+  // Map that maintains the mapping of keys in Gravitino to that in Paimon, for example, users
+  // will only need to set the configuration 'catalog-backend' in Gravitino and Gravitino will
+  // change it to `catalogType` automatically and pass it to Paimon.
+  public static final Map<String, String> GRAVITINO_CONFIG_TO_PAIMON;
+
+  static {
+    Map<String, String> map = new HashMap();
+    map.put(PaimonConstants.CATALOG_BACKEND, PaimonConstants.CATALOG_BACKEND);
+    map.put(PaimonConstants.GRAVITINO_JDBC_DRIVER, PaimonConstants.GRAVITINO_JDBC_DRIVER);
+    map.put(PaimonConstants.GRAVITINO_JDBC_USER, PaimonConstants.PAIMON_JDBC_USER);
+    map.put(PaimonConstants.GRAVITINO_JDBC_PASSWORD, PaimonConstants.PAIMON_JDBC_PASSWORD);
+    map.put(PaimonConstants.URI, PaimonConstants.URI);
+    map.put(PaimonConstants.WAREHOUSE, PaimonConstants.WAREHOUSE);
+    map.put(PaimonConstants.CATALOG_BACKEND_NAME, PaimonConstants.CATALOG_BACKEND_NAME);
+    // S3
+    map.put(S3Properties.GRAVITINO_S3_ENDPOINT, PaimonConstants.S3_ENDPOINT);
+    map.put(S3Properties.GRAVITINO_S3_ACCESS_KEY_ID, PaimonConstants.S3_ACCESS_KEY);
+    map.put(S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY, PaimonConstants.S3_SECRET_KEY);
+    // OSS
+    map.put(OSSProperties.GRAVITINO_OSS_ENDPOINT, PaimonConstants.OSS_ENDPOINT);
+    map.put(OSSProperties.GRAVITINO_OSS_ACCESS_KEY_ID, PaimonConstants.OSS_ACCESS_KEY);
+    map.put(OSSProperties.GRAVITINO_OSS_ACCESS_KEY_SECRET, PaimonConstants.OSS_SECRET_KEY);
+    GRAVITINO_CONFIG_TO_PAIMON = Collections.unmodifiableMap(map);
+  }
+
+  /**
+   * Converts Gravitino properties to Paimon catalog properties, the common transform logic shared
+   * by Spark connector, Gravitino Paimon catalog.
+   *
+   * @param gravitinoProperties a map of Gravitino configuration properties.
+   * @return a map containing Paimon catalog properties.
+   */
+  public static Map<String, String> toPaimonCatalogProperties(
+      Map<String, String> gravitinoProperties) {
+    Map<String, String> paimonProperties = new HashMap<>();
+    gravitinoProperties.forEach(
+        (key, value) -> {
+          if (GRAVITINO_CONFIG_TO_PAIMON.containsKey(key)) {
+            paimonProperties.put(GRAVITINO_CONFIG_TO_PAIMON.get(key), value);
+          }
+        });
+    return paimonProperties;
+  }
+
+  /**
+   * Get catalog backend name from Gravitino catalog properties.
+   *
+   * @param catalogProperties a map of Gravitino catalog properties.
+   * @return catalog backend name.
+   */
+  public static String getCatalogBackendName(Map<String, String> catalogProperties) {
+    String backendName = catalogProperties.get(PaimonConstants.CATALOG_BACKEND_NAME);
+    if (backendName != null) {
+      return backendName;
+    }
+
+    String catalogBackend = catalogProperties.get(PaimonConstants.CATALOG_BACKEND);
+    return Optional.ofNullable(catalogBackend)
+        .map(s -> s.toLowerCase(Locale.ROOT))
+        .orElseThrow(
+            () ->
+                new UnsupportedOperationException(
+                    String.format("Unsupported catalog backend: %s", catalogBackend)));
+  }
+}

catalogs/catalog-common/src/main/java/org/apache/gravitino/storage/S3Properties.java

@@ -32,6 +32,8 @@ public class S3Properties {
   public static final String GRAVITINO_S3_REGION = "s3-region";
   // S3 role arn
   public static final String GRAVITINO_S3_ROLE_ARN = "s3-role-arn";
+
+  public static final String GRAVITINO_S3_STS_ENDPOINT = "s3-token-service-endpoint";
   // S3 external id
   public static final String GRAVITINO_S3_EXTERNAL_ID = "s3-external-id";
 

catalogs/catalog-hadoop/build.gradle.kts

@@ -32,6 +32,7 @@ dependencies {
   implementation(project(":core")) {
     exclude(group = "*")
   }
+
   implementation(project(":common")) {
     exclude(group = "*")
   }
@@ -40,7 +41,9 @@ dependencies {
     exclude(group = "*")
   }
 
-  compileOnly(libs.guava)
+  implementation(project(":catalogs:hadoop-common")) {
+    exclude(group = "*")
+  }
 
   implementation(libs.hadoop3.common) {
     exclude("com.sun.jersey")
@@ -54,6 +57,14 @@ dependencies {
     exclude("com.sun.jersey", "jersey-servlet")
   }
 
+  implementation(libs.hadoop3.client) {
+    exclude("org.apache.hadoop", "hadoop-mapreduce-client-core")
+    exclude("org.apache.hadoop", "hadoop-mapreduce-client-jobclient")
+    exclude("org.apache.hadoop", "hadoop-yarn-api")
+    exclude("org.apache.hadoop", "hadoop-yarn-client")
+    exclude("com.squareup.okhttp", "okhttp")
+  }
+
   implementation(libs.hadoop3.hdfs) {
     exclude("com.sun.jersey")
     exclude("javax.servlet", "servlet-api")
@@ -63,38 +74,32 @@ dependencies {
     exclude("io.netty")
     exclude("org.fusesource.leveldbjni")
   }
-  implementation(libs.hadoop3.client) {
-    exclude("org.apache.hadoop", "hadoop-mapreduce-client-core")
-    exclude("org.apache.hadoop", "hadoop-mapreduce-client-jobclient")
-    exclude("org.apache.hadoop", "hadoop-yarn-api")
-    exclude("org.apache.hadoop", "hadoop-yarn-client")
-    exclude("com.squareup.okhttp", "okhttp")
-  }
 
   implementation(libs.slf4j.api)
 
-  testImplementation(project(":clients:client-java"))
-  testImplementation(project(":integration-test-common", "testArtifacts"))
-  testImplementation(project(":server"))
-  testImplementation(project(":server-common"))
+  compileOnly(libs.guava)
+
   testImplementation(project(":bundles:aws-bundle"))
   testImplementation(project(":bundles:gcp-bundle"))
   testImplementation(project(":bundles:aliyun-bundle"))
   testImplementation(project(":bundles:azure-bundle"))
-
-  testImplementation(libs.minikdc)
-  testImplementation(libs.hadoop3.minicluster)
+  testImplementation(project(":clients:client-java"))
+  testImplementation(project(":integration-test-common", "testArtifacts"))
+  testImplementation(project(":server"))
+  testImplementation(project(":server-common"))
 
   testImplementation(libs.bundles.log4j)
+  testImplementation(libs.hadoop3.gcs)
+  testImplementation(libs.hadoop3.minicluster)
+  testImplementation(libs.junit.jupiter.api)
+  testImplementation(libs.junit.jupiter.params)
+  testImplementation(libs.minikdc)
   testImplementation(libs.mockito.core)
   testImplementation(libs.mockito.inline)
   testImplementation(libs.mysql.driver)
   testImplementation(libs.postgresql.driver)
-  testImplementation(libs.junit.jupiter.api)
-  testImplementation(libs.junit.jupiter.params)
   testImplementation(libs.testcontainers)
   testImplementation(libs.testcontainers.mysql)
-  testImplementation(libs.hadoop3.gcs)
 
   testRuntimeOnly(libs.junit.jupiter.engine)
 }

catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/fs/FileSystemUtils.java

@@ -63,8 +63,8 @@ public static Map<String, FileSystemProvider> getFileSystemProviders(String file
               if (resultMap.containsKey(fileSystemProvider.scheme())) {
                 throw new UnsupportedOperationException(
                     String.format(
-                        "File system provider: '%s' with scheme '%s' already exists in the use provider list "
-                            + "Please make sure the file system provider scheme is unique.",
+                        "File system provider: '%s' with scheme '%s' already exists in the provider list,"
+                            + "please make sure the file system provider scheme is unique.",
                         fileSystemProvider.getClass().getName(), fileSystemProvider.scheme()));
               }
               resultMap.put(fileSystemProvider.scheme(), fileSystemProvider);