feat: add configurable Kerberos bootstrap timings and validation

- Introduced new Kerberos configuration options: `startupTimeoutSeconds`, `materialTimeoutSeconds`, `adminAttempts`, and `adminRetryDelaySeconds`.
- Updated container setup to use user-defined or default Kerberos timings.
- Enhanced validation to ensure positive timing-related values for Kerberos options.
- Extended TOML and programmatic configurations to support new Kerberos options.
- Updated documentation and examples to reflect added configurations.

Author: CoderYellow

core/src/main/kotlin/org/openprojectx/bigdata/test/core/BigDataTestKitOptions.kt

@@ -20,6 +20,10 @@ data class KerberosOptions(
     val clientPrincipal: String = "app_user@EXAMPLE.COM",
     val clientPassword: String = "app-user-secret",
     val users: List<KerberosUserOptions> = emptyList(),
+    val startupTimeoutSeconds: Int = 120,
+    val materialTimeoutSeconds: Int = 30,
+    val adminAttempts: Int = 30,
+    val adminRetryDelaySeconds: Int = 1,
 )
 
 data class KerberosUserOptions(

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

@@ -68,6 +68,7 @@ internal class BigDataContainerFactory(
 
     private fun kerberos(): BigDataServiceContainer {
         val kerberos = options.kerberos
+        validateKerberosTiming(kerberos)
         val servicePrincipals = buildList {
             addUserKeytab(kerberos.clientPrincipal, "/kerby/keytabs/client.keytab")
             kerberos.users.forEach { addUserKeytab(it.principal, it.keytabPath) }
@@ -91,7 +92,12 @@ internal class BigDataContainerFactory(
             .withEnv("KERBY_PA_ENC_TIMESTAMP_REQUIRED", "false")
             .withEnv("KERBY_CLIENT_PRINCIPAL", kerberos.clientPrincipal)
             .withEnv("KERBY_CLIENT_PASSWORD", kerberos.clientPassword)
-            .waitingFor(Wait.forListeningPort().withStartupTimeout(Duration.ofMinutes(2)))
+            .withEnv("KERBY_KADMIN_ATTEMPTS", kerberos.adminAttempts.toString())
+            .withEnv("KERBY_KADMIN_RETRY_DELAY_SECONDS", kerberos.adminRetryDelaySeconds.toString())
+            .waitingFor(
+                Wait.forLogMessage(".*Kerby KDC container ready\\..*", 1)
+                    .withStartupTimeout(Duration.ofSeconds(kerberos.startupTimeoutSeconds.toLong())),
+            )
         if (servicePrincipals.isNotEmpty()) {
             container.withEnv("KERBY_EXTRA_SERVICE_PRINCIPALS", servicePrincipals.joinToString(","))
         }
@@ -916,7 +922,7 @@ internal class BigDataContainerFactory(
 
     private fun waitForKerberosFiles(container: GenericContainer<*>, containerPaths: Set<String>) {
         val sources = containerPaths.map { it.replace("/kerby/", "/var/lib/kerby/") }
-        val deadline = System.nanoTime() + Duration.ofSeconds(30).toNanos()
+        val deadline = System.nanoTime() + Duration.ofSeconds(options.kerberos.materialTimeoutSeconds.toLong()).toNanos()
         while (true) {
             if (!container.isRunning) {
                 error(
@@ -945,6 +951,13 @@ internal class BigDataContainerFactory(
         }
     }
 
+    private fun validateKerberosTiming(kerberos: KerberosOptions) {
+        require(kerberos.startupTimeoutSeconds > 0) { "Kerberos startupTimeoutSeconds must be positive" }
+        require(kerberos.materialTimeoutSeconds > 0) { "Kerberos materialTimeoutSeconds must be positive" }
+        require(kerberos.adminAttempts > 0) { "Kerberos adminAttempts must be positive" }
+        require(kerberos.adminRetryDelaySeconds > 0) { "Kerberos adminRetryDelaySeconds must be positive" }
+    }
+
     private fun copyKerberosFileToContainer(container: GenericContainer<*>, containerPath: String) {
         val path = hostKerberosMaterialPath(containerPath)
         Files.createDirectories(path.parent)

doc/user-guide.adoc

@@ -706,6 +706,34 @@ class MyKerberosIntegrationTest
 
 For host-side Kafka Kerberos clients, prefer an explicit fixed `kafkaPort`. Kafka must advertise the same host and port that the client uses during the SASL/GSSAPI exchange; with a random host port the framework cannot know that advertised port before the broker starts. When Schema Registry is enabled with Kafka Kerberos, it uses Kafka's internal plaintext listener while host clients keep using the Kerberos listener.
 
+Kerberos bootstrap timing is configurable from TOML. The defaults are intentionally modest for local runs; CI can use a longer window when KDC admin startup is slower:
+
+[source,toml]
+----
+[kerberos]
+startupTimeoutSeconds = 180
+materialTimeoutSeconds = 120
+adminAttempts = 120
+adminRetryDelaySeconds = 1
+----
+
+`startupTimeoutSeconds` controls how long Testcontainers waits for the KDC image to print its ready log. `materialTimeoutSeconds` controls the post-start wait for generated keytabs and client config to be readable. `adminAttempts` and `adminRetryDelaySeconds` are passed to the Kerby image as `KERBY_KADMIN_ATTEMPTS` and `KERBY_KADMIN_RETRY_DELAY_SECONDS`.
+
+Programmatic setup exposes the same options:
+
+[source,kotlin]
+----
+BigDataTestKit.builder()
+    .withKerberos(
+        KerberosOptions(
+            enabled = true,
+            startupTimeoutSeconds = 180,
+            materialTimeoutSeconds = 120,
+            adminAttempts = 120,
+        ),
+    )
+----
+
 Default principals are defined in `BigDataTestKitOptions`:
 
 [cols="1,2",options="header"]

example/spark/src/test/resources/spark-bigdata-test-apache-hms-kerberos.toml

@@ -3,6 +3,12 @@ kerberos = true
 hiveMetastore = true
 kafkaKerberos = true
 
+[kerberos]
+startupTimeoutSeconds = 180
+materialTimeoutSeconds = 120
+adminAttempts = 120
+adminRetryDelaySeconds = 1
+
 [ports]
 kafka = 19092
 

example/spark/src/test/resources/spark-bigdata-test-cloudera-hms-kerberos.toml

@@ -3,6 +3,12 @@ kerberos = true
 clouderaHms = true
 kafkaKerberos = true
 
+[kerberos]
+startupTimeoutSeconds = 180
+materialTimeoutSeconds = 120
+adminAttempts = 120
+adminRetryDelaySeconds = 1
+
 [ports]
 kafka = 19092
 

junit5/src/main/kotlin/org/openprojectx/bigdata/test/junit5/BigDataTestConfig.kt

@@ -7,6 +7,7 @@ import org.openprojectx.bigdata.test.core.ContainerLogMode
 internal data class BigDataTestConfig(
     val images: BigDataTestImageConfig = BigDataTestImageConfig(),
     val services: BigDataTestServiceConfig = BigDataTestServiceConfig(),
+    val kerberos: BigDataTestKerberosConfig = BigDataTestKerberosConfig(),
     val tls: BigDataTestTlsConfig = BigDataTestTlsConfig(),
     val hdfsWebTls: BigDataTestHttpTlsConfig = BigDataTestHttpTlsConfig(),
     val hiveMetastoreTls: BigDataTestHttpTlsConfig = BigDataTestHttpTlsConfig(),
@@ -22,6 +23,7 @@ internal data class BigDataTestConfig(
         BigDataTestConfig(
             images = images.merge(override.images),
             services = services.merge(override.services),
+            kerberos = kerberos.merge(override.kerberos),
             tls = tls.merge(override.tls),
             hdfsWebTls = hdfsWebTls.merge(override.hdfsWebTls),
             hiveMetastoreTls = hiveMetastoreTls.merge(override.hiveMetastoreTls),
@@ -35,6 +37,21 @@ internal data class BigDataTestConfig(
         )
 }
 
+internal data class BigDataTestKerberosConfig(
+    val startupTimeoutSeconds: Int? = null,
+    val materialTimeoutSeconds: Int? = null,
+    val adminAttempts: Int? = null,
+    val adminRetryDelaySeconds: Int? = null,
+) {
+    fun merge(override: BigDataTestKerberosConfig): BigDataTestKerberosConfig =
+        BigDataTestKerberosConfig(
+            startupTimeoutSeconds = override.startupTimeoutSeconds ?: startupTimeoutSeconds,
+            materialTimeoutSeconds = override.materialTimeoutSeconds ?: materialTimeoutSeconds,
+            adminAttempts = override.adminAttempts ?: adminAttempts,
+            adminRetryDelaySeconds = override.adminRetryDelaySeconds ?: adminRetryDelaySeconds,
+        )
+}
+
 internal data class BigDataTestTlsConfig(
     val enabled: Boolean? = null,
     val caCertPath: String? = null,
@@ -188,6 +205,7 @@ internal class BigDataTestConfigLoader(
         val tables = parseTables(readText(location))
         val images = tables["images"].orEmpty()
         val services = tables["services"].orEmpty()
+        val kerberos = tables["kerberos"].orEmpty()
         val tls = tables["tls"].orEmpty()
         val hdfsWebTls = tables["hdfsWebTls"].orEmpty()
         val hiveMetastoreTls = tables["hiveMetastoreTls"].orEmpty()
@@ -226,6 +244,12 @@ internal class BigDataTestConfigLoader(
                 localStackS3 = services.boolean("localStackS3"),
                 fakeGcs = services.boolean("fakeGcs"),
             ),
+            kerberos = BigDataTestKerberosConfig(
+                startupTimeoutSeconds = kerberos.int("startupTimeoutSeconds"),
+                materialTimeoutSeconds = kerberos.int("materialTimeoutSeconds"),
+                adminAttempts = kerberos.int("adminAttempts"),
+                adminRetryDelaySeconds = kerberos.int("adminRetryDelaySeconds"),
+            ),
             tls = BigDataTestTlsConfig(
                 enabled = tls.boolean("enabled"),
                 caCertPath = tls.string("caCertPath"),
@@ -281,6 +305,7 @@ internal class BigDataTestConfigLoader(
         val knownTables = setOf(
             "images",
             "services",
+            "kerberos",
             "tls",
             "hdfsWebTls",
             "hiveMetastoreTls",

junit5/src/main/kotlin/org/openprojectx/bigdata/test/junit5/BigDataTestExtension.kt

@@ -50,6 +50,7 @@ class BigDataTestExtension : BeforeAllCallback, AfterAllCallback, ParameterResol
         val config = BigDataTestConfigLoader(context.requiredTestClass.classLoader).load(configLocations)
         val images = config.images
         val services = config.services
+        val kerberosConfig = config.kerberos
         val ports = config.ports
         val tls = config.tls
         val containerLogs = config.containerLogs
@@ -129,12 +130,17 @@ class BigDataTestExtension : BeforeAllCallback, AfterAllCallback, ParameterResol
         }
 
         if (kerberos || hdfsKerberos || hiveMetastoreKerberos || kafkaKerberos || kafkaUiKerberos) {
+            val defaultKerberos = KerberosOptions()
             builder.withKerberos(
                 KerberosOptions(
                     enabled = true,
                     image = images.kerberos ?: "ghcr.io/openprojectx/directory-kerby/kerby-kdc:latest",
                     clientPrincipal = annotation.kerberosClientPrincipal,
                     clientPassword = annotation.kerberosClientPassword,
+                    startupTimeoutSeconds = kerberosConfig.startupTimeoutSeconds ?: defaultKerberos.startupTimeoutSeconds,
+                    materialTimeoutSeconds = kerberosConfig.materialTimeoutSeconds ?: defaultKerberos.materialTimeoutSeconds,
+                    adminAttempts = kerberosConfig.adminAttempts ?: defaultKerberos.adminAttempts,
+                    adminRetryDelaySeconds = kerberosConfig.adminRetryDelaySeconds ?: defaultKerberos.adminRetryDelaySeconds,
                 ),
             )
         }