feat(extensions): support multiple HDFS JCEKS paths in S3JceksExtension

- Added `additionalHdfsPaths` to configure multiple HDFS locations for JCEKS files.
- Updated `BigDataExtensionsConfigLoader` and `BigDataExtensionsBuilder` to handle new paths via TOML and API.
- Refactored `S3JceksExtension` to generate and manage multiple HDFS JCEKS providers.
- Expanded outputs with plural paths and indexed paths for better client usage.
- Updated user documentation with examples and usage details for `additionalHdfsPaths`.
- Added unit tests to verify multi-path configurations and provider path logic.

Author: CoderYellow

doc/user-guide.adoc

@@ -1508,6 +1508,10 @@ hdfsDir = "/bigdata-test/spark"
 fileName = "s3.jceks"
 accessKeyAlias = "fs.s3a.access.key"
 secretKeyAlias = "fs.s3a.secret.key"
+additionalHdfsPaths = [
+  "/bigdata-test/app/s3.jceks",
+  "/bigdata-test/worker/s3.jceks"
+]
 
 [s3Jceks.aliases]
 "fs.s3a.encryption.algorithm" = ""
@@ -1518,12 +1522,18 @@ secretKeyAlias = "fs.s3a.secret.key"
 
 `aliases` writes additional entries into the same JCEKS file. The extension writes empty defaults for Hadoop 3.4 S3A encryption aliases because `S3AUtils.lookupPassword()` reads those options through the credential provider even when encryption is disabled. The JCEKS provider cannot store a zero-length secret, so empty alias values are stored as a single space; S3A trims the value back to blank while reading. Override them only when your test needs S3A encryption options.
 
+`additionalHdfsPaths` writes the same generated credentials to more HDFS JCEKS files. The primary file remains `hdfsDir + fileName` for backward compatibility. Use the plural output when the client should search all generated providers.
+
 Outputs:
 
 * `s3-jceks.hdfs.path`
+* `s3-jceks.hdfs.paths`
 * `s3-jceks.credential-provider.path`
+* `s3-jceks.credential-provider.paths`
+* `s3-jceks.credential-provider.path.0`
+* `s3-jceks.credential-provider.path.1`
 
-Use `s3-jceks.credential-provider.path` as `hadoop.security.credential.provider.path` or `spark.hadoop.hadoop.security.credential.provider.path`.
+Use `s3-jceks.credential-provider.path` as `hadoop.security.credential.provider.path` or `spark.hadoop.hadoop.security.credential.provider.path` for the primary generated JCEKS. Use `s3-jceks.credential-provider.paths` when `additionalHdfsPaths` is configured; it is a comma-separated Hadoop credential provider path.
 
 === Kafka Avro Seed Extension
 

extensions/src/main/kotlin/org/openprojectx/bigdata/test/extensions/config/BigDataExtensionsBuilder.kt

@@ -173,6 +173,7 @@ class S3JceksBuilder {
     var accessKeyAlias: String = "fs.s3a.access.key"
     var secretKeyAlias: String = "fs.s3a.secret.key"
     private val aliases = linkedMapOf<String, String>()
+    private val additionalHdfsPaths = mutableListOf<String>()
 
     fun alias(name: String, value: String) {
         aliases[name] = value
@@ -182,6 +183,14 @@ class S3JceksBuilder {
         aliases += values
     }
 
+    fun additionalHdfsPath(path: String) {
+        additionalHdfsPaths += path
+    }
+
+    fun additionalHdfsPaths(paths: Iterable<String>) {
+        additionalHdfsPaths += paths
+    }
+
     internal fun build(): S3JceksExtension =
         S3JceksExtension(
             id = id,
@@ -190,6 +199,7 @@ class S3JceksBuilder {
             accessKeyAlias = accessKeyAlias,
             secretKeyAlias = secretKeyAlias,
             aliases = S3JceksExtension.defaultAliases + aliases,
+            additionalHdfsPaths = additionalHdfsPaths.toList(),
         )
 }
 

extensions/src/main/kotlin/org/openprojectx/bigdata/test/extensions/config/BigDataExtensionsConfigLoader.kt

@@ -56,6 +56,7 @@ class BigDataExtensionsConfigLoader(
                     accessKeyAlias = config.string("accessKeyAlias", "fs.s3a.access.key"),
                     secretKeyAlias = config.string("secretKeyAlias", "fs.s3a.secret.key"),
                     aliases = S3JceksExtension.defaultAliases + config.stringMap("aliases"),
+                    additionalHdfsPaths = config.stringList("additionalHdfsPaths"),
                 )
             }
         }

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

@@ -12,6 +12,7 @@ data class S3JceksExtension(
     val accessKeyAlias: String = "fs.s3a.access.key",
     val secretKeyAlias: String = "fs.s3a.secret.key",
     val aliases: Map<String, String> = defaultAliases,
+    val additionalHdfsPaths: List<String> = emptyList(),
 ) : BigDataExtension {
     override val requiredServices: Set<BigDataService> = setOf(BigDataService.HDFS, BigDataService.LOCALSTACK_S3)
     override val events: Set<BigDataExtensionEvent> = setOf(BigDataExtensionEvent.AFTER_KIT_START)
@@ -20,21 +21,36 @@ data class S3JceksExtension(
         val hdfs = context.endpoint(BigDataService.HDFS)
         val s3 = context.endpoint(BigDataService.LOCALSTACK_S3)
         val kerberosProperties = context.kit.endpoints()[BigDataService.KERBEROS]?.properties.orEmpty()
-        val providerPath = HadoopCredentialProviders.hdfsJceksPath(hdfsDir, fileName)
-        HadoopCredentialProviders.createHdfsJceks(
-            hdfsProperties = hdfs.properties + kerberosProperties,
-            configDir = hdfsDir,
-            providerPath = providerPath,
-            credentials = mapOf(
-                accessKeyAlias to s3.property("aws.accessKeyId"),
-                secretKeyAlias to s3.property("aws.secretAccessKey"),
-            ) + aliases,
-        )
-        context.putOutput("$id.credential-provider.path", providerPath)
-        context.putOutput("$id.hdfs.path", "${hdfsDir.trimEnd('/')}/$fileName")
+        val credentials = mapOf(
+            accessKeyAlias to s3.property("aws.accessKeyId"),
+            secretKeyAlias to s3.property("aws.secretAccessKey"),
+        ) + aliases
+        val hdfsPaths = (listOf("${hdfsDir.trimEnd('/')}/$fileName") + additionalHdfsPaths).distinct()
+        val providerPaths = hdfsPaths.map { hdfsJceksProviderPath(it) }
+        hdfsPaths.zip(providerPaths).forEach { (hdfsPath, providerPath) ->
+            HadoopCredentialProviders.createHdfsJceks(
+                hdfsProperties = hdfs.properties + kerberosProperties,
+                configDir = hdfsPath.substringBeforeLast('/', missingDelimiterValue = "/").ifBlank { "/" },
+                providerPath = providerPath,
+                credentials = credentials,
+            )
+        }
+        context.putOutput("$id.credential-provider.path", providerPaths.first())
+        context.putOutput("$id.credential-provider.paths", providerPaths.joinToString(","))
+        context.putOutput("$id.hdfs.path", hdfsPaths.first())
+        context.putOutput("$id.hdfs.paths", hdfsPaths.joinToString(","))
+        providerPaths.forEachIndexed { index, providerPath ->
+            context.putOutput("$id.credential-provider.path.$index", providerPath)
+        }
+        hdfsPaths.forEachIndexed { index, hdfsPath ->
+            context.putOutput("$id.hdfs.path.$index", hdfsPath)
+        }
     }
 
     companion object {
+        fun hdfsJceksProviderPath(hdfsPath: String): String =
+            "jceks://hdfs/${hdfsPath.trimStart('/')}"
+
         val defaultAliases: Map<String, String> = mapOf(
             // Hadoop 3.4 S3A loads these through S3AUtils.lookupPassword().
             // Empty aliases keep credential-provider-only configurations from

extensions/src/test/kotlin/org/openprojectx/bigdata/test/extensions/hadoop/S3JceksExtensionTest.kt

@@ -13,6 +13,10 @@ class S3JceksExtensionTest {
             [s3Jceks]
             hdfsDir = "/config"
             fileName = "s3.jceks"
+            additionalHdfsPaths = [
+              "/config/app/s3.jceks",
+              "/config/worker/s3.jceks"
+            ]
 
             [s3Jceks.aliases]
             "fs.s3a.encryption.algorithm" = "SSE-S3"
@@ -23,6 +27,7 @@ class S3JceksExtensionTest {
 
         assertEquals("/config", extension.hdfsDir)
         assertEquals("s3.jceks", extension.fileName)
+        assertEquals(listOf("/config/app/s3.jceks", "/config/worker/s3.jceks"), extension.additionalHdfsPaths)
         assertEquals("SSE-S3", extension.aliases["fs.s3a.encryption.algorithm"])
         assertEquals("", extension.aliases["fs.s3a.server-side-encryption-algorithm"])
         assertEquals("", extension.aliases["fs.s3a.server-side-encryption.key"])
@@ -50,4 +55,12 @@ class S3JceksExtensionTest {
         assertTrue(defaults.containsKey("fs.s3a.encryption.key"))
         assertTrue(defaults.containsKey("fs.s3a.server-side-encryption.key"))
     }
+
+    @Test
+    fun `builds hdfs jceks provider path from hdfs path`() {
+        assertEquals(
+            "jceks://hdfs/bigdata-test/config/s3.jceks",
+            S3JceksExtension.hdfsJceksProviderPath("/bigdata-test/config/s3.jceks"),
+        )
+    }
 }