Skip to content

Commit 14c5805

Browse files
CoderYellowCoderYellow
committed
feat: add network mode customization and dependencies for HDFS testing
- Added `networkMode` support in container customizations via programmatic and TOML configurations. - Integrated `org.openprojectx.java.dns` plugin for mapping stable hostnames in local HDFS testing. - Updated `BigDataTestKit` and `BigDataContainerFactory` to support network mode settings. - Enhanced documentation with use cases for host networking and stable hostname configurations. - Added new test for validating DataNode advertisement with custom hostname and port. - Updated dependencies: added `slf4j-simple` for test logging and `java-dns` for hostname resolution.
1 parent f96ab87 commit 14c5805

9 files changed

Lines changed: 211 additions & 11 deletions

File tree

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -112,6 +112,9 @@ class BigDataTestKit private constructor(
112112
fun withContainerCustomization(service: BigDataService, options: ContainerCustomizationOptions): Builder =
113113
apply { containerCustomizations = containerCustomizations.merge(service, options) }
114114

115+
fun withContainerNetworkMode(service: BigDataService, networkMode: String): Builder =
116+
withContainerCustomization(service, ContainerCustomizationOptions(networkMode = networkMode))
117+
115118
fun withContainerEnv(service: BigDataService, name: String, value: String): Builder =
116119
withContainerCustomization(
117120
service,

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -177,6 +177,7 @@ data class BigDataHealthCheckOptions(
177177
)
178178

179179
data class ContainerCustomizationOptions(
180+
val networkMode: String? = null,
180181
val environment: Map<String, String> = emptyMap(),
181182
val files: List<ContainerFileTransferOptions> = emptyList(),
182183
val mounts: List<ContainerMountOptions> = emptyList(),
@@ -185,6 +186,7 @@ data class ContainerCustomizationOptions(
185186
) {
186187
fun merge(override: ContainerCustomizationOptions): ContainerCustomizationOptions =
187188
ContainerCustomizationOptions(
189+
networkMode = override.networkMode ?: networkMode,
188190
environment = environment + override.environment,
189191
files = files + override.files,
190192
mounts = mounts + override.mounts,

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -851,6 +851,7 @@ internal class BigDataContainerFactory(
851851

852852
private fun <T : GenericContainer<*>> applyContainerCustomizations(service: BigDataService, container: T): T {
853853
val customization = options.containerCustomizations[service] ?: return container
854+
customization.networkMode?.let { container.withNetworkMode(it) }
854855
customization.ports.forEach { container.addPort(it) }
855856
customization.environment.forEach { (name, value) -> container.withEnv(name, value) }
856857
customization.files.forEach { container.addFile(it) }

‎doc/user-guide.adoc‎

Lines changed: 34 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -277,7 +277,7 @@ Extension config has the same launcher hook: `bigdata.extensions.config` and `bi
277277

278278
Container data-driven tests can be TOML-driven because service images, ports, Kerberos flags, and extension inputs are runtime configuration. Dependency data-driven tests must be Gradle-task/classpath-driven because Spark, Hadoop, Iceberg, and vendor distributions are resolved before the test JVM starts. The Spark example uses separate resolvable runtime classpaths for Apache and Cloudera dependency lines, then pairs those classpaths with TOML service variants.
279279

280-
HDFS-specific startup settings use the `[hdfs]` table. Keep the default `dataNodeHostname = "hdfs"` for Docker-network clients. Use `localhost` only for host-side clients that connect through mapped host ports.
280+
HDFS-specific startup settings use the `[hdfs]` table. Keep the default `dataNodeHostname = "hdfs"` for Docker-network clients. Use `localhost`, or a stable hostname that resolves to `127.0.0.1`, only for host-side clients that connect through mapped host ports. Host-side clients must also use `dfs.client.use.datanode.hostname=true`; the framework exposes this endpoint property by default.
281281

282282
[source,toml]
283283
----
@@ -310,6 +310,9 @@ HADOOP_OPTS = "-Dsun.security.krb5.debug=true"
310310
311311
[containers.hdfs.ports]
312312
"9866" = 9866
313+
314+
[containers.hdfs]
315+
networkMode = "host"
313316
----
314317

315318
File sources must use one of these prefixes:
@@ -318,6 +321,8 @@ File sources must use one of these prefixes:
318321
* `file:` copies a host file path into the container.
319322
* `text:` writes inline UTF-8 text into the container.
320323

324+
`networkMode = "host"` is passed through to Testcontainers as Docker host networking. It is useful only when the local Docker runtime supports it for the target platform. The default bridge network remains the most portable option, especially when multiple test containers need to communicate through Testcontainers network aliases.
325+
321326
Mounts are useful for local debugging, but copied files are more reliable in CI because they do not depend on Docker host path sharing. Prefer copied files for certificates, keytabs, Hadoop XML, JAAS, and other test fixtures.
322327

323328
Available service names in container TOML are `kerberos`, `hdfs`, `hiveMetastore`, `hms`, `kafka`, `schemaRegistry`, `kafkaUi`, `localStackS3`, and `fakeGcs`.
@@ -329,6 +334,7 @@ Programmatic tests can use the same portable operations:
329334
val kit = BigDataTestKit.builder()
330335
.withHdfs()
331336
.withContainerEnv(BigDataService.HDFS, "HADOOP_OPTS", "-Dsun.security.krb5.debug=true")
337+
.withContainerNetworkMode(BigDataService.HDFS, "host")
332338
.withContainerFile(
333339
BigDataService.HDFS,
334340
ContainerFileTransferOptions.content(
@@ -682,9 +688,34 @@ Host ports are dynamic by default. Keep port fields set to `0` unless an externa
682688
class MyFixedPortTest
683689
----
684690

685-
For host-side HDFS clients, including Windows clients using Docker Desktop, keep the HDFS DataNode transfer port stable and set the advertised DataNode hostname to a host-reachable name. The client first connects to the NameNode, then the NameNode returns DataNode block locations. By default the framework advertises `hdfs`, which is correct for other containers on the Testcontainers network. Host-side clients usually need `hdfsDataNodeHostname = "localhost"` plus `sameHostPorts=true` or `hdfsDataNodePort = 9866`.
691+
For host-side HDFS clients, including Windows clients using Docker Desktop, keep the HDFS DataNode transfer port stable and set the advertised DataNode hostname to a host-reachable name. The client first connects to the NameNode, then the NameNode returns DataNode block locations. The client only uses the reported hostname when `dfs.client.use.datanode.hostname=true`; otherwise it may use the DataNode IP from the report and ignore the advertised hostname.
692+
693+
By default the framework advertises `hdfs`, which is correct for other containers on the Testcontainers network. Host-side clients usually need `hdfsDataNodeHostname = "localhost"` plus `sameHostPorts=true` or `hdfsDataNodePort = 9866`.
694+
695+
If both host processes and other Docker containers need to access the same HDFS instance, use an advertised hostname that is reachable from both sides. In practice that means either keeping the default `hdfs` for container-only clients, using `localhost` for host-only clients, or providing a user-managed DNS/hosts name that resolves appropriately in both environments. For JVM tests, the `org.openprojectx.java.dns` Gradle plugin can map a stable name such as `hdfs.test.local` to `127.0.0.1` without editing the OS hosts file:
696+
697+
[source,kotlin]
698+
----
699+
plugins {
700+
id("org.openprojectx.java.dns") version "0.1.1"
701+
}
702+
703+
javadns {
704+
hosts.put("hdfs.test.local", "127.0.0.1")
705+
}
706+
----
707+
708+
Then configure HDFS to advertise `hdfs.test.local` and keep the DataNode host port fixed. The Java DNS plugin affects only the client JVM's hostname resolution; the DataNode still has to register the stable hostname with the NameNode.
709+
710+
Docker host networking can also be requested for a service:
711+
712+
[source,toml]
713+
----
714+
[containers.hdfs]
715+
networkMode = "host"
716+
----
686717

687-
If both host processes and other Docker containers need to access the same HDFS instance, use an advertised hostname that is reachable from both sides. In practice that means either keeping the default `hdfs` for container-only clients, using `localhost` for host-only clients, or providing a user-managed DNS/hosts name that resolves appropriately in both environments.
718+
Host networking support depends on the Docker runtime. On Linux it maps directly to the host network namespace; on Docker Desktop it depends on the Desktop version and settings. Prefer the fixed-port plus stable-hostname approach for portable tests.
688719

689720
To bind each enabled service to the same localhost port it uses inside the container, enable `sameHostPorts`.
690721

‎extensions/build.gradle.kts‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
plugins {
22
id("buildsrc.convention.kotlin-jvm")
33
alias(libs.plugins.kotlinPluginSerialization)
4+
alias(libs.plugins.javaDns)
45
}
56

67
description = "Config-driven bigdata-test extensions for Hadoop credential providers, Kafka, and Avro"
@@ -14,4 +15,13 @@ dependencies {
1415
implementation(libs.kafkaSchemaRegistryClient)
1516
implementation(libs.avro)
1617
implementation(libs.kotlinxSerialization)
18+
19+
testImplementation(libs.junitJupiterApi)
20+
testRuntimeOnly(libs.junitJupiterEngine)
21+
testRuntimeOnly(libs.junitPlatformLauncher)
22+
testRuntimeOnly(libs.slf4jSimple)
23+
}
24+
25+
javadns {
26+
hosts.put("hdfs.test.local", "127.0.0.1")
1727
}

‎extensions/src/test/kotlin/org/openprojectx/bigdata/test/extensions/hadoop/HdfsDataNodeAdvertisementTest.kt‎

Lines changed: 108 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
1+
package org.openprojectx.bigdata.test.extensions.hadoop
2+
3+
import org.apache.hadoop.conf.Configuration
4+
import org.apache.hadoop.fs.FileSystem
5+
import org.apache.hadoop.fs.Path
6+
import org.apache.hadoop.hdfs.DistributedFileSystem
7+
import org.apache.hadoop.hdfs.protocol.DatanodeInfo
8+
import org.apache.hadoop.hdfs.protocol.HdfsConstants
9+
import org.junit.jupiter.api.Assertions.assertArrayEquals
10+
import org.junit.jupiter.api.Assertions.assertEquals
11+
import org.junit.jupiter.api.Assertions.assertTrue
12+
import org.junit.jupiter.api.Test
13+
import org.openprojectx.bigdata.test.core.BigDataService
14+
import org.openprojectx.bigdata.test.core.BigDataTestKit
15+
import org.openprojectx.bigdata.test.core.HdfsOptions
16+
import org.openprojectx.bigdata.test.core.PortBindingOptions
17+
import java.net.InetAddress
18+
import java.net.ServerSocket
19+
import java.net.URI
20+
import java.nio.charset.StandardCharsets
21+
import java.time.Duration
22+
import java.time.Instant
23+
24+
class HdfsDataNodeAdvertisementTest {
25+
private val advertisedDataNodeHost = "hdfs.test.local"
26+
private val expectedResolvedDataNodeAddress = "127.0.0.1"
27+
28+
@Test
29+
fun `hdfs client sees configured datanode advertisement`() {
30+
val dataNodePort = findAvailablePort()
31+
val kit = BigDataTestKit.builder()
32+
.withHdfs(
33+
HdfsOptions(
34+
dataNodePort = dataNodePort,
35+
dataNodeHostname = advertisedDataNodeHost,
36+
),
37+
)
38+
.withPortBindings(PortBindingOptions(hdfsDataNode = dataNodePort))
39+
.build()
40+
41+
try {
42+
kit.start()
43+
44+
val endpoint = kit.endpoint(BigDataService.HDFS)
45+
assertEquals(dataNodePort, endpoint.port("datanode"))
46+
assertEquals(advertisedDataNodeHost, endpoint.property("dfs.datanode.hostname"))
47+
assertEquals("true", endpoint.property("dfs.client.use.datanode.hostname"))
48+
49+
// val resolvedAddresses = InetAddress.getAllByName(advertisedDataNodeHost).map { it.hostAddress }
50+
// println("Resolved $advertisedDataNodeHost to $resolvedAddresses")
51+
// assertEquals(listOf(expectedResolvedDataNodeAddress), resolvedAddresses)
52+
53+
val hdfsUri = URI.create(endpoint.property("fs.defaultFS"))
54+
val configuration = Configuration().apply {
55+
endpoint.properties.forEach { (key, value) -> set(key, value) }
56+
}
57+
assertTrue(configuration.getBoolean("dfs.client.use.datanode.hostname", false))
58+
59+
FileSystem.get(hdfsUri, configuration).use { fs ->
60+
val distributedFileSystem = fs as DistributedFileSystem
61+
val dataNodes = waitForLiveDataNodes(distributedFileSystem)
62+
63+
assertEquals(1, dataNodes.size, "Expected exactly one live DataNode")
64+
val dataNode = dataNodes.single()
65+
assertEquals(advertisedDataNodeHost, dataNode.hostName)
66+
assertEquals(dataNodePort, dataNode.xferPort)
67+
assertEquals("$advertisedDataNodeHost:$dataNodePort", dataNode.getXferAddrWithHostname())
68+
assertEquals("$advertisedDataNodeHost:$dataNodePort", dataNode.getXferAddr(true))
69+
println(
70+
"NameNode reports DataNode host=${dataNode.hostName}, xferPort=${dataNode.xferPort}, " +
71+
"clientAddress=${dataNode.getXferAddr(true)}",
72+
)
73+
74+
val path = Path("/tmp/bigdata-test-datanode-advertisement.txt")
75+
val payload = "advertised-datanode-ok".toByteArray(StandardCharsets.UTF_8)
76+
fs.create(path, true).use { output -> output.write(payload) }
77+
assertTrue(fs.exists(path))
78+
fs.open(path).use { input -> assertArrayEquals(payload, input.readAllBytes()) }
79+
}
80+
} finally {
81+
kit.close()
82+
}
83+
}
84+
85+
private fun findAvailablePort(): Int =
86+
ServerSocket(0).use { socket ->
87+
socket.reuseAddress = true
88+
socket.localPort
89+
}
90+
91+
private fun waitForLiveDataNodes(distributedFileSystem: DistributedFileSystem): Array<DatanodeInfo> {
92+
val deadline = Instant.now().plus(Duration.ofSeconds(30))
93+
var lastFailure: Exception? = null
94+
95+
while (Instant.now().isBefore(deadline)) {
96+
try {
97+
val dataNodes = distributedFileSystem.getDataNodeStats(HdfsConstants.DatanodeReportType.LIVE)
98+
if (dataNodes.isNotEmpty()) return dataNodes
99+
} catch (exception: Exception) {
100+
lastFailure = exception
101+
}
102+
Thread.sleep(500)
103+
}
104+
105+
lastFailure?.let { throw it }
106+
return emptyArray()
107+
}
108+
}

‎extensions/src/test/resources/log4j2.xml‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<Configuration status="WARN" monitorInterval="30">
3+
<Properties>
4+
<Property name="CONSOLE_PATTERN">%d{yyyy-MM-dd HH:mm:ss.SSS} %highlight{%-5level}{FATAL=red, ERROR=red,
5+
WARN=yellow, INFO=green, DEBUG=cyan, TRACE=blue} [%style{%20.20t}{magenta}] %style{%logger}{cyan} -
6+
%msg%n%throwable
7+
</Property>
8+
</Properties>
9+
10+
<Appenders>
11+
<Console name="Console" target="SYSTEM_OUT">
12+
<PatternLayout pattern="${CONSOLE_PATTERN}" disableAnsi="false"/>
13+
<ThresholdFilter level="DEBUG" onMatch="ACCEPT" onMismatch="DENY"/>
14+
</Console>
15+
</Appenders>
16+
17+
<Loggers>
18+
<Logger name="org.openprojectx" level="DEBUG" additivity="false">
19+
<AppenderRef ref="Console"/>
20+
</Logger>
21+
22+
<!-- <Logger name="org.apache.spark.scheduler" level="DEBUG"/>-->
23+
<!-- <Logger name="org.apache.spark.scheduler.TaskSchedulerImpl" level="DEBUG"/>-->
24+
<!-- <Logger name="org.apache.spark.ExecutorAllocationManager" level="DEBUG"/>-->
25+
<!-- <Logger name="org.apache.spark.deploy.k8s" level="DEBUG"/>-->
26+
<!-- <Logger name="org.apache.spark.scheduler.cluster.k8s" level="DEBUG"/>-->
27+
<!-- <Logger name="org.apache.iceberg" level="DEBUG"/>-->
28+
<!-- <Logger name="org.apache.iceberg.rest" level="DEBUG"/>-->
29+
<!-- <Logger name="org.apache.hadoop.hive" level="DEBUG"/>-->
30+
31+
<Logger name="org.apache.hadoop" level="TRACE"/>
32+
33+
<Root level="INFO">
34+
<AppenderRef ref="Console"/>
35+
</Root>
36+
</Loggers>
37+
</Configuration>

‎gradle/libs.versions.toml‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,8 @@ hiveDockerTestcontainers = "0.1.4"
1313
spark = "3.5.7"
1414
hadoop = "3.4.2"
1515
kafka = "4.1.2"
16+
javaDns = "0.1.1"
17+
slf4j = "2.0.17"
1618

1719
junitJupiter = "5.14.1"
1820
junitPlatform = "1.14.1"
@@ -57,6 +59,7 @@ avro = { module = "org.apache.avro:avro", version.ref = "avro" }
5759
junitJupiterApi = { module = "org.junit.jupiter:junit-jupiter-api", version.ref = "junitJupiter" }
5860
junitJupiterEngine = { module = "org.junit.jupiter:junit-jupiter-engine", version.ref = "junitJupiter" }
5961
junitPlatformLauncher = { module = "org.junit.platform:junit-platform-launcher", version.ref = "junitPlatform" }
62+
slf4jSimple = { module = "org.slf4j:slf4j-simple", version.ref = "slf4j" }
6063
bouncycastlePkix = { module = "org.bouncycastle:bcpkix-jdk18on", version.ref = "bouncycastle" }
6164
bouncycastleProvider = { module = "org.bouncycastle:bcprov-jdk18on", version.ref = "bouncycastle" }
6265

@@ -66,3 +69,4 @@ kotlinxEcosystem = ["kotlinxDatetime", "kotlinxSerialization", "kotlinxCoroutine
6669

6770
[plugins]
6871
kotlinPluginSerialization = { id = "org.jetbrains.kotlin.plugin.serialization", version.ref = "kotlin" }
72+
javaDns = { id = "org.openprojectx.java.dns", version.ref = "javaDns" }

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

Lines changed: 12 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -468,16 +468,20 @@ internal class BigDataTestConfigLoader(
468468
tables.forEach { (table, values) ->
469469
if (!table.startsWith("containers.")) return@forEach
470470
val parts = table.split('.')
471-
require(parts.size == 3) {
472-
"TOML table '$table' must use containers.<service>.<env|files|mounts|ports>"
471+
require(parts.size == 2 || parts.size == 3) {
472+
"TOML table '$table' must use containers.<service> or containers.<service>.<env|files|mounts|ports>"
473473
}
474474
val service = parseService(parts[1], table)
475-
val options = when (parts[2]) {
476-
"env" -> ContainerCustomizationOptions(environment = values.stringMap(table))
477-
"files" -> ContainerCustomizationOptions(files = values.containerFiles())
478-
"mounts" -> ContainerCustomizationOptions(mounts = values.containerMounts())
479-
"ports" -> ContainerCustomizationOptions(ports = values.containerPorts(table))
480-
else -> error("TOML table '$table' must end with env, files, mounts, or ports")
475+
val options = if (parts.size == 2) {
476+
ContainerCustomizationOptions(networkMode = values.string("networkMode"))
477+
} else {
478+
when (parts[2]) {
479+
"env" -> ContainerCustomizationOptions(environment = values.stringMap(table))
480+
"files" -> ContainerCustomizationOptions(files = values.containerFiles())
481+
"mounts" -> ContainerCustomizationOptions(mounts = values.containerMounts())
482+
"ports" -> ContainerCustomizationOptions(ports = values.containerPorts(table))
483+
else -> error("TOML table '$table' must end with env, files, mounts, or ports")
484+
}
481485
}
482486
customizations[service] = customizations[service]?.merge(options) ?: options
483487
}

0 commit comments

Comments
 (0)