feat: improve HMS object store integration and external table validation

- Injected Hadoop filesystem configuration for HMS when using LocalStack S3 or fake GCS.
- Updated Spark example to validate Hive external Parquet tables with enhanced metadata checks.
- Refactored container initialization to ensure consistent order of service startup.
- Added support for validating table location and input format in HMS metadata verification.

Author: CoderYellow

README.md

@@ -159,6 +159,7 @@ class MyFixedPortTest
 
 Available JUnit port fields are `kerberosKdcPort`, `hdfsNameNodePort`, `hdfsWebPort`, `hiveMetastorePort`, `kafkaPort`, `schemaRegistryPort`, `kafkaUiPort`, `localStackS3Port`, and `fakeGcsPort`. Endpoint properties still use the actual mapped host ports returned by Testcontainers.
 
+When Hive Metastore is started with LocalStack S3 or fake GCS, the kit also injects server-side Hadoop filesystem configuration into HMS. S3 external table DDL can validate `s3a://` locations against the internal LocalStack endpoint. GCS config is also injected, but HMS still needs a GCS connector jar in the HMS image before it can validate `gs://` locations.
 
 ## Spring Boot
 
@@ -203,7 +204,7 @@ GRADLE_USER_HOME=/data/.gradle ./gradlew :example:spring:bootRun --args='--sprin
 
 The JUnit examples in `example/junit` are annotated with `@Disabled`; remove that annotation from an example class to start the configured stack.
 
-The Spark example in `example/spark` creates a `SparkSession` from `BigDataTestKit` endpoints, starts HDFS, Hive Metastore, Kafka, Schema Registry, LocalStack S3, and fake GCS, then uses `extensions` config to create the S3 JCEKS file on HDFS and produce Avro records to Kafka. HDFS is used as the config store for the JCEKS file; S3 and GCS are object-store Iceberg smoke checks through Hadoop catalogs. The example also creates an HMS-backed Iceberg table and a Hive metastore Parquet table stored on S3, then verifies the HMS database/table metadata through `HiveMetaStoreClient`.
+The Spark example in `example/spark` creates a `SparkSession` from `BigDataTestKit` endpoints, starts HDFS, Hive Metastore, Kafka, Schema Registry, LocalStack S3, and fake GCS, then uses `extensions` config to create the S3 JCEKS file on HDFS and produce Avro records to Kafka. HDFS is used as the config store for the JCEKS file; S3 and GCS are object-store Iceberg smoke checks through Hadoop catalogs. The example also creates an HMS-backed Iceberg table and a Hive external Parquet table stored on S3, then verifies the HMS database/table location and format metadata through `HiveMetaStoreClient`.
 
 ```bash
 GRADLE_USER_HOME=/data/.gradle ./gradlew :example:spark:test

core/src/main/kotlin/org/openprojectx/bigdata/test/core/container/BigDataContainerFactory.kt

@@ -37,14 +37,14 @@ internal class BigDataContainerFactory(
         val containers = mutableListOf<BigDataServiceContainer>()
         if (kerberosRequired()) containers += kerberos()
         if (options.hdfs.enabled) containers += hdfs()
+        if (options.localStackS3.enabled) containers += localStackS3()
+        if (options.fakeGcs.enabled) containers += fakeGcs()
         if (options.hiveMetastore.enabled) containers += hiveMetastore()
         if (options.kafka.enabled) {
             containers += kafka()
             if (options.kafka.schemaRegistryEnabled) containers += schemaRegistry()
             if (options.kafka.kafkaUiEnabled) containers += kafkaUi()
         }
-        if (options.localStackS3.enabled) containers += localStackS3()
-        if (options.fakeGcs.enabled) containers += fakeGcs()
         return containers
     }
 
@@ -179,6 +179,9 @@ internal class BigDataContainerFactory(
                 .withEnv("HMS_CONF_HADOOP_SECURITY_AUTHENTICATION", "kerberos")
         }
 
+        hiveMetastoreObjectStoreConfiguration().forEach { (key, value) ->
+            container.withEnv("HMS_CONF_${encodeConfigKey(key)}", value)
+        }
         hive.extraConfiguration.forEach { (key, value) ->
             container.withEnv("HMS_CONF_${encodeConfigKey(key)}", value)
         }
@@ -388,6 +391,33 @@ internal class BigDataContainerFactory(
     }
 
 
+    private fun hiveMetastoreObjectStoreConfiguration(): Map<String, String> =
+        buildMap {
+            if (options.localStackS3.enabled) {
+                put("fs.s3a.endpoint", "http://localstack:4566")
+                put("fs.s3a.endpoint.region", "us-east-1")
+                put("fs.s3a.aws.credentials.provider", "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider")
+                put("fs.s3a.access.key", "test")
+                put("fs.s3a.secret.key", "test")
+                put("fs.s3a.path.style.access", "true")
+                put("fs.s3a.connection.ssl.enabled", "false")
+                put("fs.s3a.change.detection.mode", "none")
+            }
+            if (options.fakeGcs.enabled) {
+                put("fs.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem")
+                put("fs.AbstractFileSystem.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS")
+                put("fs.gs.project.id", "bigdata-test")
+                put("fs.gs.storage.root.url", "http://fake-gcs:4443/")
+                put("fs.gs.storage.service.path", "storage/v1/")
+                put("fs.gs.client.type", "HTTP_API_CLIENT")
+                put("fs.gs.auth.type", "UNAUTHENTICATED")
+                put("fs.gs.create.items.conflict.check.enable", "false")
+                put("fs.gs.implicit.dir.repair.enable", "false")
+                put("fs.gs.hierarchical.namespace.folders.enable", "false")
+            }
+        }
+
+
     private fun <T : GenericContainer<*>> attachLogs(name: String, container: T): T {
         when (options.containerLogs.mode) {
             ContainerLogMode.NONE -> Unit

example/spark/src/test/kotlin/org/openprojectx/bigdata/test/example/spark/SparkBigDataTestExample.kt

@@ -73,12 +73,13 @@ class SparkBigDataTestExample {
                 database = "hms_demo_$runId",
                 table = "events_hms",
             )
-            assertHiveS3ParquetTable(
+            assertHiveExternalParquetTable(
                 spark = spark,
                 hiveMetastoreUri = hiveMetastore.property("hive.metastore.uris"),
                 database = "hive_s3_demo_$runId",
                 table = "events_parquet_s3",
                 location = "s3a://$s3Bucket/hive-parquet/events_parquet_s3",
+                storageName = "hive-s3-parquet",
             )
         }
     }
@@ -195,49 +196,52 @@ class SparkBigDataTestExample {
         }
     }
 
-    private fun assertHiveS3ParquetTable(
+    private fun assertHiveExternalParquetTable(
         spark: SparkSession,
         hiveMetastoreUri: String,
         database: String,
         table: String,
         location: String,
+        storageName: String,
     ) {
         val identifier = "$database.$table"
         spark.sql("CREATE DATABASE IF NOT EXISTS $database")
         spark.sql(
             """
-            CREATE TABLE $identifier (
+            CREATE EXTERNAL TABLE $identifier (
                 id INT,
                 name STRING,
                 storage STRING
-            ) USING PARQUET
+            ) STORED AS PARQUET
             LOCATION '$location'
             """.trimIndent(),
         )
-        spark.sql("INSERT INTO $identifier VALUES (1, 'alpha', 'hive-s3-parquet'), (2, 'beta', 'hive-s3-parquet')")
-        val count = spark.table(identifier).where("storage = 'hive-s3-parquet'").count()
-        check(count == 2L) { "Expected two Hive S3 Parquet rows in $identifier" }
-        assertS3ParquetFiles(spark, location)
-        assertHiveMetastoreParquetS3Table(hiveMetastoreUri, database, table)
+        spark.sql("INSERT INTO $identifier VALUES (1, 'alpha', '$storageName'), (2, 'beta', '$storageName')")
+        val count = spark.table(identifier).where("storage = '$storageName'").count()
+        check(count == 2L) { "Expected two Hive external Parquet rows in $identifier" }
+        assertParquetFiles(spark, location)
+        assertHiveMetastoreExternalParquetTable(hiveMetastoreUri, database, table, location)
     }
 
-    private fun assertS3ParquetFiles(spark: SparkSession, location: String) {
+    private fun assertParquetFiles(spark: SparkSession, location: String) {
         val files = org.apache.hadoop.fs.FileSystem.get(URI.create(location), spark.sparkContext().hadoopConfiguration())
             .use { fs -> fs.listStatus(org.apache.hadoop.fs.Path(location)).map { it.path.name } }
         check(files.any { it.endsWith(".parquet") }) { "Expected Parquet files under $location, got $files" }
     }
 
-    private fun assertHiveMetastoreParquetS3Table(hiveMetastoreUri: String, database: String, table: String) {
+    private fun assertHiveMetastoreExternalParquetTable(hiveMetastoreUri: String, database: String, table: String, location: String) {
         val conf = HiveConf()
         conf.setVar(HiveConf.ConfVars.METASTOREURIS, hiveMetastoreUri)
         val client = HiveMetaStoreClient(conf)
         try {
             val hmsTable = client.getTable(database, table)
             check(hmsTable.dbName == database) { "Expected HMS table $database.$table, got ${hmsTable.dbName}.${hmsTable.tableName}" }
             check(hmsTable.tableName == table) { "Expected HMS table $database.$table, got ${hmsTable.dbName}.${hmsTable.tableName}" }
-            val provider = hmsTable.parameters["spark.sql.sources.provider"] ?: hmsTable.parameters["provider"]
-            check(provider.equals("parquet", ignoreCase = true) || hmsTable.sd.inputFormat.contains("Parquet")) {
-                "Expected HMS table $database.$table to be Parquet, parameters=${hmsTable.parameters}, inputFormat=${hmsTable.sd.inputFormat}"
+            check(hmsTable.sd.location == location) {
+                "Expected HMS table $database.$table location $location, got ${hmsTable.sd.location}"
+            }
+            check(hmsTable.sd.inputFormat.contains("Parquet")) {
+                "Expected HMS table $database.$table to be Parquet, inputFormat=${hmsTable.sd.inputFormat}"
             }
         } finally {
             client.close()