feat(extensions): add support for custom JCEKS aliases in S3JceksExtension

- Introduced `aliases` parameter to define custom JCEKS entries, alongside default Hadoop 3.4 S3A encryption options.
- Updated `BigDataExtensionsConfigLoader` and `BigDataExtensionsBuilder` to support `aliases` from TOML configurations.
- Ensured empty aliases are stored as single spaces to maintain compatibility with S3AUtils.
- Added comprehensive tests for default and custom alias handling logic.
- Expanded user documentation with configuration and usage details for `aliases`.

Author: CoderYellow

doc/user-guide.adoc

@@ -1481,8 +1481,16 @@ hdfsDir = "/bigdata-test/spark"
 fileName = "s3.jceks"
 accessKeyAlias = "fs.s3a.access.key"
 secretKeyAlias = "fs.s3a.secret.key"
+
+[s3Jceks.aliases]
+"fs.s3a.encryption.algorithm" = ""
+"fs.s3a.server-side-encryption-algorithm" = ""
+"fs.s3a.encryption.key" = ""
+"fs.s3a.server-side-encryption.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.
+
 Outputs:
 
 * `s3-jceks.hdfs.path`

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

@@ -3,6 +3,10 @@ enabled = true
 hdfsDir = "/bigdata-test/spark"
 fileName = "s3.jceks"
 
+[s3Jceks.aliases]
+"fs.s3a.encryption.algorithm" = ""
+"fs.s3a.server-side-encryption-algorithm" = ""
+
 [kafkaAvro]
 enabled = true
 

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

@@ -164,6 +164,15 @@ class S3JceksBuilder {
     var fileName: String = "s3.jceks"
     var accessKeyAlias: String = "fs.s3a.access.key"
     var secretKeyAlias: String = "fs.s3a.secret.key"
+    private val aliases = linkedMapOf<String, String>()
+
+    fun alias(name: String, value: String) {
+        aliases[name] = value
+    }
+
+    fun aliases(values: Map<String, String>) {
+        aliases += values
+    }
 
     internal fun build(): S3JceksExtension =
         S3JceksExtension(
@@ -172,6 +181,7 @@ class S3JceksBuilder {
             fileName = fileName,
             accessKeyAlias = accessKeyAlias,
             secretKeyAlias = secretKeyAlias,
+            aliases = S3JceksExtension.defaultAliases + aliases,
         )
 }
 

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

@@ -55,6 +55,7 @@ class BigDataExtensionsConfigLoader(
                     fileName = config.string("fileName", "s3.jceks"),
                     accessKeyAlias = config.string("accessKeyAlias", "fs.s3a.access.key"),
                     secretKeyAlias = config.string("secretKeyAlias", "fs.s3a.secret.key"),
+                    aliases = S3JceksExtension.defaultAliases + config.stringMap("aliases"),
                 )
             }
         }
@@ -163,6 +164,24 @@ class BigDataExtensionsConfigLoader(
             )
         }.orEmpty()
 
+    private fun JsonObject.stringMap(name: String): Map<String, String> =
+        this[name]?.jsonObject?.flattenStringMap(errorPath = name).orEmpty()
+
+    private fun JsonObject.flattenStringMap(prefix: String = "", errorPath: String): Map<String, String> =
+        flatMap { (key, value) ->
+            val outputKey = if (prefix.isBlank()) key else "$prefix.$key"
+            val fullErrorPath = "$errorPath.$key"
+            when (value) {
+                is JsonObject -> value.flattenStringMap(prefix = outputKey, errorPath = fullErrorPath).toList()
+                else -> listOf(
+                    outputKey to (
+                        value.jsonPrimitive.contentOrNull
+                            ?: error("Extension config field '$fullErrorPath' must be a string")
+                    ),
+                )
+            }
+        }.toMap()
+
     private fun JsonObject.string(name: String): String =
         this[name]?.jsonPrimitive?.contentOrNull ?: error("Missing required extension config field '$name'")
 

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

@@ -44,7 +44,7 @@ object HadoopCredentialProviders {
             if (provider.getCredentialEntry(alias) != null) {
                 provider.deleteCredentialEntry(alias)
             }
-            provider.createCredentialEntry(alias, value.toCharArray())
+            provider.createCredentialEntry(alias, value.ifEmpty { " " }.toCharArray())
         }
         provider.flush()
     }

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

@@ -11,6 +11,7 @@ data class S3JceksExtension(
     val fileName: String = "s3.jceks",
     val accessKeyAlias: String = "fs.s3a.access.key",
     val secretKeyAlias: String = "fs.s3a.secret.key",
+    val aliases: Map<String, String> = defaultAliases,
 ) : BigDataExtension {
     override val requiredServices: Set<BigDataService> = setOf(BigDataService.HDFS, BigDataService.LOCALSTACK_S3)
     override val events: Set<BigDataExtensionEvent> = setOf(BigDataExtensionEvent.AFTER_KIT_START)
@@ -27,9 +28,21 @@ data class S3JceksExtension(
             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")
     }
+
+    companion object {
+        val defaultAliases: Map<String, String> = mapOf(
+            // Hadoop 3.4 S3A loads these through S3AUtils.lookupPassword().
+            // Empty aliases keep credential-provider-only configurations from
+            // failing when encryption is not enabled.
+            "fs.s3a.encryption.algorithm" to "",
+            "fs.s3a.server-side-encryption-algorithm" to "",
+            "fs.s3a.encryption.key" to "",
+            "fs.s3a.server-side-encryption.key" to "",
+        )
+    }
 }

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

@@ -0,0 +1,53 @@
+package org.openprojectx.bigdata.test.extensions.hadoop
+
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+import org.openprojectx.bigdata.test.extensions.config.BigDataExtensionsConfigLoader
+
+class S3JceksExtensionTest {
+    @Test
+    fun `loads default and custom jceks aliases from toml`() {
+        val extension = BigDataExtensionsConfigLoader().load(
+            """
+            [s3Jceks]
+            hdfsDir = "/config"
+            fileName = "s3.jceks"
+
+            [s3Jceks.aliases]
+            "fs.s3a.encryption.algorithm" = "SSE-S3"
+            "fs.s3a.encryption.key" = ""
+            "custom.alias" = "custom-value"
+            """.trimIndent(),
+        ).single() as S3JceksExtension
+
+        assertEquals("/config", extension.hdfsDir)
+        assertEquals("s3.jceks", extension.fileName)
+        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"])
+        assertEquals("custom-value", extension.aliases["custom.alias"])
+    }
+
+    @Test
+    fun `loads unquoted dotted jceks aliases from toml`() {
+        val extension = BigDataExtensionsConfigLoader().load(
+            """
+            [s3Jceks.aliases]
+            fs.s3a.encryption.algorithm = "SSE-KMS"
+            """.trimIndent(),
+        ).single() as S3JceksExtension
+
+        assertEquals("SSE-KMS", extension.aliases["fs.s3a.encryption.algorithm"])
+    }
+
+    @Test
+    fun `default jceks aliases include s3a encryption options`() {
+        val defaults = S3JceksExtension.defaultAliases
+
+        assertTrue(defaults.containsKey("fs.s3a.encryption.algorithm"))
+        assertTrue(defaults.containsKey("fs.s3a.server-side-encryption-algorithm"))
+        assertTrue(defaults.containsKey("fs.s3a.encryption.key"))
+        assertTrue(defaults.containsKey("fs.s3a.server-side-encryption.key"))
+    }
+}