feat: enhance Spark example and improve HDFS container startup

- Updated Spark example with detailed connectivity checks for HDFS and Hive Metastore.
- Added `hdfsStartupCommand` for improved HDFS container initialization.
- Configured Gradle tests to include necessary JVM arguments.
- Adjusted container log mode to file for Spark tests.

Author: CoderYellow

README.md

@@ -132,7 +132,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` shows a JUnit test that creates a `SparkSession` from `BigDataTestKit` endpoints and configures HDFS, Hive Metastore, Kafka, S3A, and fake GCS settings. It is also disabled by default because it starts the full container 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, verifies HDFS and Hive Metastore connectivity, and configures Spark Kafka/S3/GCS settings:
 
 ```bash
 GRADLE_USER_HOME=/data/.gradle ./gradlew :example:spark:test

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

@@ -112,7 +112,8 @@ internal class BigDataContainerFactory(
             .withNetwork(network)
             .withNetworkAliases("hdfs", "hdfs.example.com")
             .withExposedPorts(hdfs.nameNodePort, hdfs.webPort)
-            .waitingFor(Wait.forListeningPort().withStartupTimeout(Duration.ofMinutes(3)))
+            .withCommand("sh", "-lc", hdfsStartupCommand(hdfs.nameNodePort, hdfs.webPort))
+            .waitingFor(Wait.forHttp("/").forPort(hdfs.webPort).withStartupTimeout(Duration.ofMinutes(3)))
         if (hdfs.kerberos.enabled) {
             mountKerberos(container)
             container
@@ -400,6 +401,31 @@ internal class BigDataContainerFactory(
     private fun sanitizeLogName(name: String): String =
         name.replace(Regex("[^A-Za-z0-9._-]"), "_")
 
+    private fun hdfsStartupCommand(nameNodePort: Int, webPort: Int): String =
+        """
+        set -eu
+        cat > "${'$'}HADOOP_CONF_DIR/core-site.xml" <<EOF
+        <configuration>
+          <property><name>fs.defaultFS</name><value>hdfs://hdfs:$nameNodePort</value></property>
+        </configuration>
+        EOF
+        cat > "${'$'}HADOOP_CONF_DIR/hdfs-site.xml" <<EOF
+        <configuration>
+          <property><name>dfs.replication</name><value>1</value></property>
+          <property><name>dfs.permissions.enabled</name><value>false</value></property>
+          <property><name>dfs.namenode.name.dir</name><value>file:///tmp/hadoop-name</value></property>
+          <property><name>dfs.datanode.data.dir</name><value>file:///tmp/hadoop-data</value></property>
+          <property><name>dfs.namenode.rpc-bind-host</name><value>0.0.0.0</value></property>
+          <property><name>dfs.namenode.http-address</name><value>0.0.0.0:$webPort</value></property>
+          <property><name>dfs.namenode.http-bind-host</name><value>0.0.0.0</value></property>
+        </configuration>
+        EOF
+        hdfs namenode -format -force -nonInteractive
+        hdfs namenode &
+        hdfs datanode &
+        wait
+        """.trimIndent()
+
     private fun encodeConfigKey(key: String): String =
         key.lowercase(Locale.ROOT)
             .replace("-", "__")

example/spark/build.gradle.kts

@@ -14,3 +14,10 @@ dependencies {
     testRuntimeOnly(libs.junitJupiterEngine)
     testRuntimeOnly(libs.junitPlatformLauncher)
 }
+
+
+tasks.withType<Test>().configureEach {
+    jvmArgs(
+        "--add-exports=java.base/sun.nio.ch=ALL-UNNAMED",
+    )
+}

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

@@ -1,22 +1,24 @@
 package org.openprojectx.bigdata.test.example.spark
 
+import org.apache.hadoop.fs.FileSystem
+import org.apache.hadoop.fs.Path
 import org.apache.spark.sql.SparkSession
-import org.junit.jupiter.api.Disabled
+import java.net.URI
+import java.nio.file.Files
 import org.junit.jupiter.api.Test
 import org.openprojectx.bigdata.test.core.BigDataService
 import org.openprojectx.bigdata.test.core.BigDataTestKit
 import org.openprojectx.bigdata.test.core.ContainerLogMode
 import org.openprojectx.bigdata.test.junit5.BigDataTest
 
-@Disabled("Example only. Remove @Disabled to start Spark against the configured Testcontainers stack.")
 @BigDataTest(
     hdfs = true,
     hiveMetastore = true,
     kafka = true,
     schemaRegistry = true,
     localStackS3 = true,
     fakeGcs = true,
-    containerLogMode = ContainerLogMode.STDOUT
+    containerLogMode = ContainerLogMode.FILE
 )
 class SparkBigDataTestExample {
     @Test
@@ -28,11 +30,12 @@ class SparkBigDataTestExample {
         val s3 = kit.endpoint(BigDataService.LOCALSTACK_S3)
         val gcs = kit.endpoint(BigDataService.FAKE_GCS)
 
+        val warehouseDir = Files.createTempDirectory("bigdata-test-spark-warehouse-").toUri().toString()
         val spark = SparkSession.builder()
             .appName("bigdata-test-spark-example")
             .master("local[2]")
             .config("spark.ui.enabled", "false")
-            .config("spark.sql.warehouse.dir", "${hdfs.property("fs.defaultFS")}/user/hive/warehouse")
+            .config("spark.sql.warehouse.dir", warehouseDir)
             .config("hive.metastore.uris", hiveMetastore.property("hive.metastore.uris"))
             .config("spark.hadoop.fs.defaultFS", hdfs.property("fs.defaultFS"))
             .config("spark.hadoop.fs.s3a.endpoint", s3.property("aws.endpoint-url.s3"))
@@ -45,6 +48,9 @@ class SparkBigDataTestExample {
             .getOrCreate()
 
         spark.use { session ->
+            FileSystem.get(URI.create(hdfs.property("fs.defaultFS")), session.sparkContext().hadoopConfiguration())
+                .use { fs -> check(fs.exists(Path("/"))) }
+
             session.sql("CREATE DATABASE IF NOT EXISTS bigdata_test_example")
             session.sql("CREATE TABLE IF NOT EXISTS bigdata_test_example.spark_smoke (id INT, name STRING) USING parquet")
             session.sql("INSERT INTO bigdata_test_example.spark_smoke VALUES (1, 'spark')")