feat: add HDFS DataNode customization and CLI health checks

- Introduced `hdfsDataNodeHostname` and `hdfsDataNodePort` options for fine-tuned HDFS configuration.
- Added support for container-level customizations, such as environment variables, file copies, ports, and mounts via programmatic and TOML configurations.
- Enabled optional CLI-based health checks for all services, including HDFS, Kafka, and Hive Metastore.
- Extended documentation with detailed examples and guidance for DataNode configuration, container customizations, and health checks.
- Updated Spark example with additional Jackson version handling for Cloudera dependencies.

Author: CoderYellow

README.md

@@ -78,6 +78,26 @@ domain = "localhost"
 
 The endpoint properties then return HTTPS URLs and JVM truststore settings such as `javax.net.ssl.trustStore`.
 
+For image-specific troubleshooting, containers can be customized from TOML:
+
+```toml
+[containers.hdfs.env]
+HADOOP_OPTS = "-Dsun.security.krb5.debug=true"
+
+[containers.hdfs.files]
+"/tmp/message.txt" = "text:created by bigdata-test"
+```
+
+Programmatic tests can also call `customizeContainer(...)` for last-resort Testcontainers access.
+
+Optional CLI health checks can run after container startup:
+
+```toml
+[healthChecks]
+hdfs = "cli"
+localStackS3 = "cli"
+```
+
 ## Run Examples
 
 Use the shared Gradle home when running this repository locally:

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

@@ -17,8 +17,14 @@ enum class BigDataService(
         ),
     ),
     HDFS(
-        defaultPorts = mapOf("namenode" to 8020, "web" to 9870, "web-tls" to 443),
-        endpointProperties = setOf("fs.defaultFS", "spring.hadoop.fs-uri", "dfs.namenode.https-address"),
+        defaultPorts = mapOf("namenode" to 8020, "datanode" to 9866, "web" to 9870, "web-tls" to 443),
+        endpointProperties = setOf(
+            "fs.defaultFS",
+            "spring.hadoop.fs-uri",
+            "dfs.client.use.datanode.hostname",
+            "dfs.datanode.hostname",
+            "dfs.namenode.https-address",
+        ),
     ),
     HIVE_METASTORE(
         defaultPorts = mapOf("thrift" to 9083),

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

@@ -1,7 +1,9 @@
 package org.openprojectx.bigdata.test.core
 
 import org.openprojectx.bigdata.test.core.container.BigDataContainerFactory
+import org.testcontainers.containers.GenericContainer
 import org.testcontainers.lifecycle.Startable
+import java.util.function.Consumer
 
 class BigDataTestKit private constructor(
     private val options: BigDataTestKitOptions,
@@ -16,7 +18,9 @@ class BigDataTestKit private constructor(
         serviceContainers.forEach {
             it.container.start()
             it.afterStart()
-            endpoints[it.service] = it.endpoint()
+            val endpoint = it.endpoint()
+            endpoints[it.service] = endpoint
+            factory.healthCheck(it.service, it.container, endpoint)
         }
         started = true
     }
@@ -57,6 +61,8 @@ class BigDataTestKit private constructor(
         private var fakeGcs = ObjectStoreOptions(image = "fsouza/fake-gcs-server:1.54")
         private var portBindings = PortBindingOptions()
         private var containerLogs = ContainerLogOptions()
+        private var containerCustomizations = emptyMap<BigDataService, ContainerCustomizationOptions>()
+        private var healthChecks = emptyMap<BigDataService, BigDataHealthCheckOptions>()
 
         fun withKerberos(options: KerberosOptions = KerberosOptions(enabled = true)): Builder =
             apply { kerberos = options.copy(enabled = true) }
@@ -103,6 +109,43 @@ class BigDataTestKit private constructor(
         fun withContainerLogsToDirectory(directory: String = "build/bigdata-test-container-logs"): Builder =
             withContainerLogs(ContainerLogOptions(mode = ContainerLogMode.FILE, directory = directory))
 
+        fun withContainerCustomization(service: BigDataService, options: ContainerCustomizationOptions): Builder =
+            apply { containerCustomizations = containerCustomizations.merge(service, options) }
+
+        fun withContainerEnv(service: BigDataService, name: String, value: String): Builder =
+            withContainerCustomization(
+                service,
+                ContainerCustomizationOptions(environment = mapOf(name to value)),
+            )
+
+        fun withContainerFile(service: BigDataService, file: ContainerFileTransferOptions): Builder =
+            withContainerCustomization(service, ContainerCustomizationOptions(files = listOf(file)))
+
+        fun withContainerMount(service: BigDataService, mount: ContainerMountOptions): Builder =
+            withContainerCustomization(service, ContainerCustomizationOptions(mounts = listOf(mount)))
+
+        fun withContainerPort(service: BigDataService, port: ContainerPortOptions): Builder =
+            withContainerCustomization(service, ContainerCustomizationOptions(ports = listOf(port)))
+
+        fun customizeContainer(service: BigDataService, customizer: BigDataContainerCustomizer): Builder =
+            withContainerCustomization(service, ContainerCustomizationOptions(customizers = listOf(customizer)))
+
+        fun customizeContainer(service: BigDataService, customizer: Consumer<GenericContainer<*>>): Builder =
+            customizeContainer(service, BigDataContainerCustomizer { customizer.accept(it) })
+
+        fun withHealthCheck(service: BigDataService, options: BigDataHealthCheckOptions): Builder =
+            apply { healthChecks = healthChecks + (service to options) }
+
+        fun withCliHealthCheck(service: BigDataService): Builder =
+            withHealthCheck(service, BigDataHealthCheckOptions(mode = BigDataHealthCheckMode.CLI))
+
+        fun withCliHealthChecks(): Builder =
+            apply {
+                BigDataService.entries.forEach {
+                    withCliHealthCheck(it)
+                }
+            }
+
         fun build(): BigDataTestKit =
             BigDataTestKit(
                 BigDataTestKitOptions(
@@ -115,7 +158,15 @@ class BigDataTestKit private constructor(
                     fakeGcs = fakeGcs,
                     portBindings = portBindings,
                     containerLogs = containerLogs,
+                    containerCustomizations = containerCustomizations,
+                    healthChecks = healthChecks,
                 ),
             )
+
+        private fun Map<BigDataService, ContainerCustomizationOptions>.merge(
+            service: BigDataService,
+            options: ContainerCustomizationOptions,
+        ): Map<BigDataService, ContainerCustomizationOptions> =
+            this + (service to (this[service]?.merge(options) ?: options))
     }
 }

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

@@ -1,5 +1,7 @@
 package org.openprojectx.bigdata.test.core
 
+import org.testcontainers.containers.GenericContainer
+
 data class BigDataTestKitOptions(
     val kerberos: KerberosOptions = KerberosOptions(),
     val tls: TlsOptions = TlsOptions(),
@@ -10,6 +12,8 @@ data class BigDataTestKitOptions(
     val fakeGcs: ObjectStoreOptions = ObjectStoreOptions(image = "fsouza/fake-gcs-server:1.54"),
     val portBindings: PortBindingOptions = PortBindingOptions(),
     val containerLogs: ContainerLogOptions = ContainerLogOptions(),
+    val containerCustomizations: Map<BigDataService, ContainerCustomizationOptions> = emptyMap(),
+    val healthChecks: Map<BigDataService, BigDataHealthCheckOptions> = emptyMap(),
 )
 
 data class KerberosOptions(
@@ -61,6 +65,8 @@ data class HdfsOptions(
     val enabled: Boolean = false,
     val image: String = "apache/hadoop:3.5.0",
     val nameNodePort: Int = 8020,
+    val dataNodePort: Int = 9866,
+    val dataNodeHostname: String = "hdfs",
     val webPort: Int = 9870,
     val webTls: HttpTlsOptions = HttpTlsOptions(),
     val kerberos: KerberosAuthOptions = KerberosAuthOptions(
@@ -124,6 +130,7 @@ data class PortBindingOptions(
     val sameHostPorts: Boolean = false,
     val kerberosKdc: Int = 0,
     val hdfsNameNode: Int = 0,
+    val hdfsDataNode: Int = 0,
     val hdfsWeb: Int = 0,
     val hdfsWebTls: Int = 0,
     val hiveMetastore: Int = 0,
@@ -157,3 +164,88 @@ data class ContainerLogOptions(
     val mode: ContainerLogMode = ContainerLogMode.NONE,
     val directory: String = "build/bigdata-test-container-logs",
 )
+
+enum class BigDataHealthCheckMode {
+    NONE,
+    BASIC,
+    CLI,
+}
+
+data class BigDataHealthCheckOptions(
+    val mode: BigDataHealthCheckMode = BigDataHealthCheckMode.BASIC,
+    val timeoutSeconds: Long = 60,
+)
+
+data class ContainerCustomizationOptions(
+    val environment: Map<String, String> = emptyMap(),
+    val files: List<ContainerFileTransferOptions> = emptyList(),
+    val mounts: List<ContainerMountOptions> = emptyList(),
+    val ports: List<ContainerPortOptions> = emptyList(),
+    val customizers: List<BigDataContainerCustomizer> = emptyList(),
+) {
+    fun merge(override: ContainerCustomizationOptions): ContainerCustomizationOptions =
+        ContainerCustomizationOptions(
+            environment = environment + override.environment,
+            files = files + override.files,
+            mounts = mounts + override.mounts,
+            ports = ports + override.ports,
+            customizers = customizers + override.customizers,
+        )
+}
+
+data class ContainerFileTransferOptions(
+    val containerPath: String,
+    val hostPath: String? = null,
+    val content: ByteArray? = null,
+    val fileMode: Int? = null,
+) {
+    init {
+        require(containerPath.isNotBlank()) { "Container file path must not be blank" }
+        require((hostPath == null) != (content == null)) {
+            "Exactly one of hostPath or content must be set for container file transfer"
+        }
+    }
+
+    companion object {
+        @JvmStatic
+        fun hostPath(hostPath: String, containerPath: String, fileMode: Int? = null): ContainerFileTransferOptions =
+            ContainerFileTransferOptions(containerPath = containerPath, hostPath = hostPath, fileMode = fileMode)
+
+        @JvmStatic
+        fun content(content: String, containerPath: String, fileMode: Int? = null): ContainerFileTransferOptions =
+            ContainerFileTransferOptions(
+                containerPath = containerPath,
+                content = content.toByteArray(Charsets.UTF_8),
+                fileMode = fileMode,
+            )
+
+        @JvmStatic
+        fun content(content: ByteArray, containerPath: String, fileMode: Int? = null): ContainerFileTransferOptions =
+            ContainerFileTransferOptions(containerPath = containerPath, content = content, fileMode = fileMode)
+    }
+}
+
+data class ContainerMountOptions(
+    val hostPath: String,
+    val containerPath: String,
+    val readOnly: Boolean = true,
+) {
+    init {
+        require(hostPath.isNotBlank()) { "Host mount path must not be blank" }
+        require(containerPath.isNotBlank()) { "Container mount path must not be blank" }
+    }
+}
+
+data class ContainerPortOptions(
+    val containerPort: Int,
+    val hostPort: Int = 0,
+) {
+    init {
+        require(containerPort > 0) { "Container port must be positive" }
+        require(hostPort >= 0) { "Host port must be 0 for random binding or a positive fixed port" }
+    }
+}
+
+fun interface BigDataContainerCustomizer {
+    fun customize(container: GenericContainer<*>)
+}