[#5410] Improvement(iceberg-reset): Make IT IcebergRestKerberosHiveCatalogIT works in deploy mode.

iceberg/iceberg-rest-server/build.gradle.kts

@@ -91,6 +91,7 @@ dependencies {
   testImplementation(libs.sqlite.jdbc)
   testImplementation(libs.slf4j.api)
   testImplementation(libs.testcontainers)
+  testImplementation(libs.hadoop2.common)
 
   testRuntimeOnly(libs.junit.jupiter.engine)
 }

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRESTServiceBaseIT.java

@@ -53,7 +53,7 @@
 public abstract class IcebergRESTServiceBaseIT {
 
   public static final Logger LOG = LoggerFactory.getLogger(IcebergRESTServiceBaseIT.class);
-  private SparkSession sparkSession;
+  protected SparkSession sparkSession;
   protected IcebergCatalogBackend catalogType = IcebergCatalogBackend.MEMORY;
   private IcebergRESTServerManager icebergRESTServerManager;
 
@@ -121,14 +121,14 @@ private void registerIcebergCatalogConfig() {
     LOG.info("Iceberg REST service config registered, {}", StringUtils.join(icebergConfigs));
   }
 
-  private int getServerPort() {
+  protected int getServerPort() {
     JettyServerConfig jettyServerConfig =
         JettyServerConfig.fromConfig(
             icebergRESTServerManager.getServerConfig(), IcebergConfig.ICEBERG_CONFIG_PREFIX);
     return jettyServerConfig.getHttpPort();
   }
 
-  private void initSparkEnv() {
+  protected void initSparkEnv() {
     int port = getServerPort();
     LOG.info("Iceberg REST server port:{}", port);
     String icebergRESTUri = String.format("http://127.0.0.1:%d/iceberg/", port);

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRESTServiceIT.java

@@ -70,7 +70,7 @@ private void purgeNameSpace(String namespace) {
     sql("DROP database " + namespace);
   }
 
-  private void purgeAllIcebergTestNamespaces() {
+  protected void purgeAllIcebergTestNamespaces() {
     List<Object[]> databases =
         sql(String.format("SHOW DATABASES like '%s*'", ICEBERG_REST_NS_PREFIX));
     Set<String> databasesString = convertToStringSet(databases, 0);

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRestKerberosHiveCatalogIT.java

@@ -23,21 +23,24 @@
 import java.nio.file.Files;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.Objects;
 import org.apache.commons.io.FileUtils;
 import org.apache.gravitino.iceberg.common.IcebergCatalogBackend;
 import org.apache.gravitino.iceberg.common.IcebergConfig;
+import org.apache.gravitino.iceberg.common.authentication.kerberos.KerberosClient;
 import org.apache.gravitino.integration.test.container.HiveContainer;
 import org.apache.gravitino.integration.test.util.GravitinoITUtils;
-import org.apache.gravitino.integration.test.util.ITUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.spark.SparkConf;
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.TestInstance;
 import org.junit.jupiter.api.TestInstance.Lifecycle;
-import org.junit.jupiter.api.condition.EnabledIf;
+import org.testcontainers.shaded.com.google.common.collect.ImmutableMap;
 
 @Tag("gravitino-docker-test")
 @TestInstance(Lifecycle.PER_CLASS)
-@EnabledIf("isEmbedded")
 public class IcebergRestKerberosHiveCatalogIT extends IcebergRESTHiveCatalogIT {
 
   private static final String HIVE_METASTORE_CLIENT_PRINCIPAL = "cli@HADOOPKRB";
@@ -149,12 +152,50 @@ Map<String, String> getCatalogConfig() {
     return configMap;
   }
 
-  private static boolean isEmbedded() {
-    String mode =
-        System.getProperty(ITUtils.TEST_MODE) == null
-            ? ITUtils.EMBEDDED_TEST_MODE
-            : System.getProperty(ITUtils.TEST_MODE);
+  protected void initSparkEnv() {
+    int port = getServerPort();
+    LOG.info("Iceberg REST server port:{}", port);
+    String icebergRESTUri = String.format("http://127.0.0.1:%d/iceberg/", port);
+    SparkConf sparkConf =
+        new SparkConf()
+            .set(
+                "spark.sql.extensions",
+                "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
+            .set("spark.sql.catalog.rest", "org.apache.iceberg.spark.SparkCatalog")
+            .set("spark.sql.catalog.rest.type", "rest")
+            .set("spark.sql.catalog.rest.uri", icebergRESTUri)
+            .set("spark.locality.wait.node", "0");
+
+    if (supportsCredentialVending()) {
+      sparkConf.set(
+          "spark.sql.catalog.rest.header.X-Iceberg-Access-Delegation", "vended-credentials");
+    }
+
+    // Login kerberos and use the users to execute the spark job.
+    try {
+      Configuration configuration = new Configuration();
+      configuration.set("hadoop.security.authentication", "kerberos");
+      KerberosClient kerberosClient =
+          new KerberosClient(
+              ImmutableMap.of(
+                  "authentication.kerberos.principal",
+                  HIVE_METASTORE_CLIENT_PRINCIPAL,
+                  "authentication.kerberos.keytab-uri",
+                  tempDir + HIVE_METASTORE_CLIENT_KEYTAB),
+              configuration);
+
+      kerberosClient.login(tempDir + HIVE_METASTORE_CLIENT_KEYTAB);
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+
+    System.setProperty("spark.hadoop.hadoop.security.authentication", "kerberos");
+    sparkSession = SparkSession.builder().master("local[1]").config(sparkConf).getOrCreate();
+  }
 
-    return Objects.equals(mode, ITUtils.EMBEDDED_TEST_MODE);
+  @AfterAll
+  void cleanup() {
+    purgeAllIcebergTestNamespaces();
+    UserGroupInformation.reset();
   }
 }