Skip to content

Commit ac633d0

Browse files
CoderYellowCoderYellow
committed
feat: add container logging support and Spark example
1 parent 33fdd3d commit ac633d0

13 files changed

Lines changed: 246 additions & 35 deletions

File tree

‎README.md‎

Lines changed: 48 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ Composable Testcontainers-based fixtures for local big-data integration tests.
1010
- `bigdata-test-spring-boot-starter`: starter that brings in the auto-configuration
1111
- `example:spring`: Spring Boot local-development example
1212
- `example:junit`: JUnit 5 integration-test examples
13+
- `example:spark`: Spark + JUnit 5 example that wires Spark to the container endpoints
1314

1415
## Core Usage
1516

@@ -27,6 +28,24 @@ kit.use {
2728
}
2829
```
2930

31+
Container logs are disabled by default. Enable them when a container fails to start or you need service-side troubleshooting output:
32+
33+
```kotlin
34+
val kit = BigDataTestKit.builder()
35+
.withKafka()
36+
.withContainerLogsToStdout()
37+
.build()
38+
```
39+
40+
or write one file per service:
41+
42+
```kotlin
43+
val kit = BigDataTestKit.builder()
44+
.withHiveMetastore()
45+
.withContainerLogsToDirectory("build/container-logs")
46+
.build()
47+
```
48+
3049
## JUnit 5
3150

3251
```kotlin
@@ -39,11 +58,13 @@ class MyIntegrationTest {
3958
}
4059
```
4160

42-
Kerberos can be enabled per service:
61+
Kerberos can be enabled per service. Hadoop/HDFS, Hive Metastore, Kafka, Schema Registry, and Kafka UI expose dedicated Kerberos switches; set `kerberos = true` to start the shared KDC and then enable auth for the services that should use it.
4362

4463
```kotlin
4564
@BigDataTest(
4665
kerberos = true,
66+
hdfs = true,
67+
hdfsKerberos = true,
4768
hiveMetastore = true,
4869
hiveMetastoreKerberos = true,
4970
kafka = true,
@@ -54,6 +75,20 @@ Kerberos can be enabled per service:
5475
class MyKerberosIntegrationTest
5576
```
5677

78+
JUnit tests can also route container logs to the main test process console or to files:
79+
80+
```kotlin
81+
@BigDataTest(
82+
kafka = true,
83+
schemaRegistry = true,
84+
containerLogMode = ContainerLogMode.FILE,
85+
containerLogDirectory = "build/container-logs",
86+
)
87+
class MyTroubleshootingTest
88+
```
89+
90+
`ContainerLogMode.STDOUT` prefixes each line with the service name. `ContainerLogMode.FILE` writes files such as `kafka.log`, `schema-registry.log`, `hive-metastore.log`, and `hive-metastore-postgres.log`.
91+
5792
## Spring Boot
5893

5994
Add the starter and enable the kit in local-development or test configuration:
@@ -75,6 +110,12 @@ bigdata:
75110
76111
The auto-configuration exposes a started `BigDataTestKit` bean and closes it with the application context.
77112

113+
## Dependency Management
114+
115+
The build uses the Testcontainers BOM `org.testcontainers:testcontainers-bom:2.0.4`. In this repository it is applied once from the root `subprojects` block, so individual modules and examples should depend on Testcontainers modules without repeating the BOM.
116+
117+
For external consumers, import the same BOM in your own dependency-management setup before adding `bigdata-test` and any direct Testcontainers dependencies. With Testcontainers 2.x, module coordinates use the `testcontainers-` prefix, for example `org.testcontainers:testcontainers-junit-jupiter` and `org.testcontainers:testcontainers-postgresql`.
118+
78119
## Examples
79120

80121
Run the Spring example without starting containers:
@@ -90,3 +131,9 @@ GRADLE_USER_HOME=/data/.gradle ./gradlew :example:spring:bootRun --args='--sprin
90131
```
91132

92133
The JUnit examples in `example/junit` are annotated with `@Disabled`; remove that annotation from an example class to start the configured stack.
134+
135+
The Spark example in `example/spark` shows a JUnit test that creates a `SparkSession` from `BigDataTestKit` endpoints and configures HDFS, Hive Metastore, Kafka, S3A, and fake GCS settings. It is also disabled by default because it starts the full container stack:
136+
137+
```bash
138+
GRADLE_USER_HOME=/data/.gradle ./gradlew :example:spark:test
139+
```

‎build.gradle.kts‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,17 @@ allprojects {
1616
subprojects {
1717
tasks.register<DependencyReportTask>("allDependencies") {}
1818

19+
val testcontainersBomConfigurations = setOf(
20+
"api",
21+
"implementation",
22+
"testImplementation",
23+
"kapt",
24+
"kaptTest",
25+
)
26+
configurations.matching { it.name in testcontainersBomConfigurations }.all {
27+
project.dependencies.add(name, project.dependencies.enforcedPlatform(libs.testcontainersBom))
28+
}
29+
1930
if (path.startsWith(":example:")) {
2031
return@subprojects
2132
}

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

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,7 @@ class BigDataTestKit private constructor(
5252
private var kafka = KafkaOptions()
5353
private var localStackS3 = ObjectStoreOptions()
5454
private var fakeGcs = ObjectStoreOptions(image = "fsouza/fake-gcs-server:1.54")
55+
private var containerLogs = ContainerLogOptions()
5556

5657
fun withKerberos(options: KerberosOptions = KerberosOptions(enabled = true)): Builder =
5758
apply { kerberos = options.copy(enabled = true) }
@@ -74,6 +75,15 @@ class BigDataTestKit private constructor(
7475
fun withFakeGcs(options: ObjectStoreOptions = ObjectStoreOptions(enabled = true, image = "fsouza/fake-gcs-server:1.54")): Builder =
7576
apply { fakeGcs = options.copy(enabled = true) }
7677

78+
fun withContainerLogs(options: ContainerLogOptions): Builder =
79+
apply { containerLogs = options }
80+
81+
fun withContainerLogsToStdout(): Builder =
82+
withContainerLogs(ContainerLogOptions(mode = ContainerLogMode.STDOUT))
83+
84+
fun withContainerLogsToDirectory(directory: String = "build/bigdata-test-container-logs"): Builder =
85+
withContainerLogs(ContainerLogOptions(mode = ContainerLogMode.FILE, directory = directory))
86+
7787
fun build(): BigDataTestKit =
7888
BigDataTestKit(
7989
BigDataTestKitOptions(
@@ -84,6 +94,7 @@ class BigDataTestKit private constructor(
8494
kafka = kafka,
8595
localStackS3 = localStackS3,
8696
fakeGcs = fakeGcs,
97+
containerLogs = containerLogs,
8798
),
8899
)
89100
}

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

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ data class BigDataTestKitOptions(
88
val kafka: KafkaOptions = KafkaOptions(),
99
val localStackS3: ObjectStoreOptions = ObjectStoreOptions(),
1010
val fakeGcs: ObjectStoreOptions = ObjectStoreOptions(image = "fsouza/fake-gcs-server:1.54"),
11+
val containerLogs: ContainerLogOptions = ContainerLogOptions(),
1112
)
1213

1314
data class KerberosOptions(
@@ -84,3 +85,15 @@ data class ObjectStoreOptions(
8485
val enabled: Boolean = false,
8586
val image: String = "localstack/localstack:4.14.0",
8687
)
88+
89+
90+
enum class ContainerLogMode {
91+
NONE,
92+
STDOUT,
93+
FILE,
94+
}
95+
96+
data class ContainerLogOptions(
97+
val mode: ContainerLogMode = ContainerLogMode.NONE,
98+
val directory: String = "build/bigdata-test-container-logs",
99+
)

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

Lines changed: 57 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -3,16 +3,23 @@ package org.openprojectx.bigdata.test.core.container
33
import org.openprojectx.bigdata.test.core.BigDataEndpoint
44
import org.openprojectx.bigdata.test.core.BigDataService
55
import org.openprojectx.bigdata.test.core.BigDataTestKitOptions
6+
import org.openprojectx.bigdata.test.core.ContainerLogMode
67
import org.openprojectx.bigdata.test.core.KerberosAuthOptions
78
import org.testcontainers.containers.BindMode
9+
import org.testcontainers.containers.GenericContainer
810
import org.testcontainers.containers.Network
911
import org.testcontainers.containers.PostgreSQLContainer
1012
import org.testcontainers.containers.wait.strategy.Wait
13+
import org.testcontainers.containers.output.OutputFrame
1114
import org.testcontainers.images.builder.Transferable
1215
import org.testcontainers.lifecycle.Startable
1316
import org.testcontainers.utility.DockerImageName
17+
import java.io.OutputStreamWriter
18+
import java.io.Closeable
1419
import java.nio.file.Files
20+
import java.nio.charset.StandardCharsets
1521
import java.nio.file.Path
22+
import java.nio.file.StandardOpenOption
1623
import java.time.Duration
1724
import java.util.Locale
1825

@@ -21,6 +28,7 @@ internal class BigDataContainerFactory(
2128
) : AutoCloseable {
2229
private val network = Network.newNetwork()
2330
private val supportContainers = mutableListOf<Startable>()
31+
private val logConsumers = mutableListOf<Closeable>()
2432
private val kerberosDir: Path? = if (kerberosRequired()) Files.createTempDirectory("bigdata-test-kerberos-") else null
2533

2634
fun create(): List<BigDataServiceContainer> {
@@ -40,6 +48,7 @@ internal class BigDataContainerFactory(
4048

4149
override fun close() {
4250
supportContainers.asReversed().forEach { it.stop() }
51+
logConsumers.asReversed().forEach { it.close() }
4352
network.close()
4453
}
4554

@@ -82,7 +91,7 @@ internal class BigDataContainerFactory(
8291
container.withEnv("KERBY_EXTRA_SERVICE_PRINCIPALS", servicePrincipals.joinToString(","))
8392
}
8493

85-
return BigDataServiceContainer(BigDataService.KERBEROS, container) {
94+
return BigDataServiceContainer(BigDataService.KERBEROS, attachLogs("kerberos", container)) {
8695
BigDataEndpoint(
8796
service = BigDataService.KERBEROS,
8897
host = container.host,
@@ -114,7 +123,7 @@ internal class BigDataContainerFactory(
114123
.withEnv("HDFS_NAMENODE_KEYTAB_FILE", hdfs.kerberos.keytabPath)
115124
}
116125

117-
return BigDataServiceContainer(BigDataService.HDFS, container) {
126+
return BigDataServiceContainer(BigDataService.HDFS, attachLogs("hdfs", container)) {
118127
val nameNode = "${container.host}:${container.getMappedPort(hdfs.nameNodePort)}"
119128
BigDataEndpoint(
120129
service = BigDataService.HDFS,
@@ -141,6 +150,7 @@ internal class BigDataContainerFactory(
141150
.withNetwork(network)
142151
.withNetworkAliases("hms-postgres")
143152
supportContainers += postgres
153+
attachLogs("hive-metastore-postgres", postgres)
144154

145155
val container = GenericBigDataContainer(hive.image)
146156
.withNetwork(network)
@@ -169,7 +179,7 @@ internal class BigDataContainerFactory(
169179
container.withEnv("HMS_CONF_${encodeConfigKey(key)}", value)
170180
}
171181

172-
return BigDataServiceContainer(BigDataService.HIVE_METASTORE, container) {
182+
return BigDataServiceContainer(BigDataService.HIVE_METASTORE, attachLogs("hive-metastore", container)) {
173183
val thriftUri = "thrift://${container.host}:${container.getMappedPort(9083)}"
174184
BigDataEndpoint(
175185
service = BigDataService.HIVE_METASTORE,
@@ -221,7 +231,7 @@ internal class BigDataContainerFactory(
221231
)
222232
}
223233

224-
return BigDataServiceContainer(BigDataService.KAFKA, container) {
234+
return BigDataServiceContainer(BigDataService.KAFKA, attachLogs("kafka", container)) {
225235
val bootstrapServers = "${container.host}:${container.getMappedPort(9092)}"
226236
BigDataEndpoint(
227237
service = BigDataService.KAFKA,
@@ -259,7 +269,7 @@ internal class BigDataContainerFactory(
259269
.withEnv("SCHEMA_REGISTRY_OPTS", "-Djava.security.krb5.conf=/kerby/client/krb5.conf")
260270
}
261271

262-
return BigDataServiceContainer(BigDataService.SCHEMA_REGISTRY, container) {
272+
return BigDataServiceContainer(BigDataService.SCHEMA_REGISTRY, attachLogs("schema-registry", container)) {
263273
val url = "http://${container.host}:${container.getMappedPort(8085)}"
264274
BigDataEndpoint(
265275
service = BigDataService.SCHEMA_REGISTRY,
@@ -291,7 +301,7 @@ internal class BigDataContainerFactory(
291301
.withEnv("KAFKA_CLUSTERS_0_PROPERTIES_SASL_JAAS_CONFIG", inlineJaas(kafka.kafkaUiKerberos))
292302
}
293303

294-
return BigDataServiceContainer(BigDataService.KAFKA_UI, container) {
304+
return BigDataServiceContainer(BigDataService.KAFKA_UI, attachLogs("kafka-ui", container)) {
295305
val url = "http://${container.host}:${container.getMappedPort(8080)}"
296306
BigDataEndpoint(
297307
service = BigDataService.KAFKA_UI,
@@ -311,7 +321,7 @@ internal class BigDataContainerFactory(
311321
.withEnv("SERVICES", "s3")
312322
.waitingFor(Wait.forHttp("/_localstack/health").forStatusCode(200).withStartupTimeout(Duration.ofMinutes(3)))
313323

314-
return BigDataServiceContainer(BigDataService.LOCALSTACK_S3, container) {
324+
return BigDataServiceContainer(BigDataService.LOCALSTACK_S3, attachLogs("localstack-s3", container)) {
315325
val endpoint = "http://${container.host}:${container.getMappedPort(4566)}"
316326
BigDataEndpoint(
317327
service = BigDataService.LOCALSTACK_S3,
@@ -337,7 +347,7 @@ internal class BigDataContainerFactory(
337347
.withCommand("-scheme", "http", "-port", "4443")
338348
.waitingFor(Wait.forHttp("/storage/v1/b").forStatusCode(200).withStartupTimeout(Duration.ofMinutes(2)))
339349

340-
return BigDataServiceContainer(BigDataService.FAKE_GCS, container) {
350+
return BigDataServiceContainer(BigDataService.FAKE_GCS, attachLogs("fake-gcs", container)) {
341351
val endpoint = "http://${container.host}:${container.getMappedPort(4443)}"
342352
BigDataEndpoint(
343353
service = BigDataService.FAKE_GCS,
@@ -351,6 +361,45 @@ internal class BigDataContainerFactory(
351361
}
352362
}
353363

364+
365+
private fun <T : GenericContainer<*>> attachLogs(name: String, container: T): T {
366+
when (options.containerLogs.mode) {
367+
ContainerLogMode.NONE -> Unit
368+
ContainerLogMode.STDOUT -> container.withLogConsumer { frame -> writeConsoleFrame(name, frame) }
369+
ContainerLogMode.FILE -> {
370+
val logDir = Files.createDirectories(Path.of(options.containerLogs.directory))
371+
val writer = OutputStreamWriter(
372+
Files.newOutputStream(
373+
logDir.resolve("${sanitizeLogName(name)}.log"),
374+
StandardOpenOption.CREATE,
375+
StandardOpenOption.APPEND,
376+
),
377+
StandardCharsets.UTF_8,
378+
)
379+
logConsumers += writer
380+
container.withLogConsumer { frame -> writeFileFrame(writer, frame) }
381+
}
382+
}
383+
return container
384+
}
385+
386+
private fun writeConsoleFrame(name: String, frame: OutputFrame) {
387+
val text = frame.utf8String.removeSuffix("\n")
388+
if (text.isEmpty()) return
389+
val stream = if (frame.type == OutputFrame.OutputType.STDERR) System.err else System.out
390+
text.lineSequence().forEach { line -> stream.println("[$name] $line") }
391+
}
392+
393+
private fun writeFileFrame(writer: OutputStreamWriter, frame: OutputFrame) {
394+
synchronized(writer) {
395+
writer.write(frame.utf8String)
396+
writer.flush()
397+
}
398+
}
399+
400+
private fun sanitizeLogName(name: String): String =
401+
name.replace(Regex("[^A-Za-z0-9._-]"), "_")
402+
354403
private fun encodeConfigKey(key: String): String =
355404
key.lowercase(Locale.ROOT)
356405
.replace("-", "__")

‎example/spark/build.gradle.kts‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
plugins {
2+
id("buildsrc.convention.kotlin-jvm")
3+
}
4+
5+
description = "Spark JUnit 5 example for bigdata-test"
6+
7+
dependencies {
8+
testImplementation(project(":junit5"))
9+
testImplementation(libs.sparkSql)
10+
testImplementation(libs.sparkHive)
11+
testImplementation(libs.sparkSqlKafka)
12+
testImplementation(libs.hadoopAws)
13+
testImplementation(libs.junitJupiterApi)
14+
testRuntimeOnly(libs.junitJupiterEngine)
15+
testRuntimeOnly(libs.junitPlatformLauncher)
16+
}

0 commit comments

Comments
 (0)