feat: refactor HDFS credential management and enhance local testing

- Replaced `hdfsUri` with `hdfsProperties` in S3 JCEKS Extension for more flexible HDFS configurations.
- Introduced `hdfsConfiguration` utility method to streamline HDFS property handling.
- Added JavaDNS plugin to support stable hostname resolution for local HDFS testing.
- Updated TOML configuration and user guide to reflect new HDFS property handling and DataNode customization options.

Author: CoderYellow

doc/user-guide.adoc

@@ -1117,6 +1117,7 @@ Outputs:
 === S3 JCEKS Extension
 
 The `s3Jceks` extension creates a Hadoop credential provider file in HDFS and stores LocalStack S3 credentials in it.
+It builds its Hadoop client configuration from the active HDFS endpoint properties, including `fs.defaultFS`, `dfs.client.use.datanode.hostname`, `dfs.datanode.hostname`, and any HDFS TLS properties exposed by the test kit.
 
 [source,toml]
 ----

example/spark/build.gradle.kts

@@ -8,6 +8,7 @@ plugins {
     id("buildsrc.convention.kotlin-jvm")
     id("org.openprojectx.spark.platform") version "0.1.41"
     id("org.openprojectx.hadoop-native-loader") version "0.1.1"
+    alias(libs.plugins.javaDns)
 
 }
 
@@ -112,6 +113,10 @@ tasks.withType<Test>().configureEach {
     )
 }
 
+javadns {
+    hosts.put("hdfs", "127.0.0.1")
+}
+
 tasks.named<Test>("test") {
     useSparkRuntimeClasspath(clouderaSparkRuntimeClasspath, "cloudera")
 }

example/spark/src/test/resources/spark-bigdata-test-common.toml

@@ -17,5 +17,8 @@ schemaRegistry = true
 localStackS3 = true
 fakeGcs = true
 
+[ports]
+hdfsDataNode = 9866
+
 [containerLogs]
 mode = "FILE"

extensions/build.gradle.kts

@@ -24,4 +24,6 @@ dependencies {
 
 javadns {
     hosts.put("hdfs.test.local", "127.0.0.1")
+    hosts.put("hdfs", "127.0.0.1")
+
 }

extensions/src/main/kotlin/org/openprojectx/bigdata/test/extensions/hadoop/HadoopCredentialProviders.kt

@@ -16,9 +16,25 @@ object HadoopCredentialProviders {
         providerPath: String,
         credentials: Map<String, String>,
     ) {
-        val conf = Configuration(false)
-        conf.set("fs.defaultFS", hdfsUri)
+        createHdfsJceks(
+            hdfsProperties = mapOf("fs.defaultFS" to hdfsUri),
+            configDir = configDir,
+            providerPath = providerPath,
+            credentials = credentials,
+        )
+    }
+
+    fun createHdfsJceks(
+        hdfsProperties: Map<String, String>,
+        configDir: String,
+        providerPath: String,
+        credentials: Map<String, String>,
+    ) {
+        val conf = hdfsConfiguration(hdfsProperties)
         conf.set(CredentialProviderFactory.CREDENTIAL_PROVIDER_PATH, providerPath)
+        val hdfsUri = requireNotNull(hdfsProperties["fs.defaultFS"]?.takeIf { it.isNotBlank() }) {
+            "HDFS properties must include non-blank fs.defaultFS"
+        }
         FileSystem.get(URI.create(hdfsUri), conf).use { fs -> fs.mkdirs(Path(configDir)) }
 
         val provider = CredentialProviderFactory.getProviders(conf).single()
@@ -32,8 +48,16 @@ object HadoopCredentialProviders {
     }
 
     fun exists(hdfsUri: String, path: String): Boolean {
-        val conf = Configuration(false)
-        conf.set("fs.defaultFS", hdfsUri)
+        val conf = hdfsConfiguration(mapOf("fs.defaultFS" to hdfsUri))
         return FileSystem.get(URI.create(hdfsUri), conf).use { fs -> fs.exists(Path(path)) }
     }
+
+    private fun hdfsConfiguration(properties: Map<String, String>): Configuration =
+        Configuration(false).apply {
+            properties.forEach { (key, value) ->
+                if (value.isNotBlank()) {
+                    set(key, value)
+                }
+            }
+        }
 }

extensions/src/main/kotlin/org/openprojectx/bigdata/test/extensions/hadoop/S3JceksExtension.kt

@@ -20,7 +20,7 @@ data class S3JceksExtension(
         val s3 = context.endpoint(BigDataService.LOCALSTACK_S3)
         val providerPath = HadoopCredentialProviders.hdfsJceksPath(hdfsDir, fileName)
         HadoopCredentialProviders.createHdfsJceks(
-            hdfsUri = hdfs.property("fs.defaultFS"),
+            hdfsProperties = hdfs.properties,
             configDir = hdfsDir,
             providerPath = providerPath,
             credentials = mapOf(