Skip to content

Commit 921574b

Browse files
CoderYellowCoderYellow
committed
feat: add addons module for Hadoop, Kafka, and Avro test helpers
- Introduced `addons` module for optional test utilities requiring heavier dependencies. - Added `HadoopCredentialProviders` for managing HDFS-backed JCEKS files. - Added `KafkaAvroProducers` for Kafka topic creation and Avro record production. - Updated Spark example to leverage `addons` for S3 JCEKS and Avro-Kafka integration. - Simplified dependencies in the Spark example by consolidating libraries into `addons`.
1 parent 1cb2979 commit 921574b

6 files changed

Lines changed: 177 additions & 73 deletions

File tree

‎README.md‎

Lines changed: 32 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ Composable Testcontainers-based fixtures for local big-data integration tests.
66

77
- `core`: container builder and endpoint/property model
88
- `junit5`: `@BigDataTest` extension for JUnit 5 tests
9+
- `addons`: optional Hadoop, Kafka, and Avro helpers for integration-test setup
910
- `bigdata-test-spring-boot-autoconfigure`: Spring Boot auto-configuration
1011
- `bigdata-test-spring-boot-starter`: starter that brings in the auto-configuration
1112
- `example:spring`: Spring Boot local-development example
@@ -46,6 +47,36 @@ val kit = BigDataTestKit.builder()
4647
.build()
4748
```
4849

50+
## Addons
51+
52+
`core` and `junit5` intentionally stay clean of Hadoop, Spark, and storage-client dependencies. Put optional test setup helpers behind `addons` when they need heavier runtime libraries. The current addon module includes:
53+
54+
- `HadoopCredentialProviders`: creates HDFS-backed JCEKS files, useful for S3A credentials such as `fs.s3a.access.key` and `fs.s3a.secret.key`.
55+
- `KafkaAvroProducers`: creates a topic when needed and produces Avro records through Schema Registry.
56+
57+
```kotlin
58+
val configDir = "/bigdata-test/spark/$runId"
59+
val providerPath = HadoopCredentialProviders.hdfsJceksPath(configDir, "s3.jceks")
60+
61+
HadoopCredentialProviders.createHdfsJceks(
62+
hdfsUri = hdfs.property("fs.defaultFS"),
63+
configDir = configDir,
64+
providerPath = providerPath,
65+
credentials = mapOf(
66+
"fs.s3a.access.key" to s3.property("aws.accessKeyId"),
67+
"fs.s3a.secret.key" to s3.property("aws.secretAccessKey"),
68+
),
69+
)
70+
71+
KafkaAvroProducers.produce(
72+
bootstrapServers = kafka.property("bootstrap.servers"),
73+
schemaRegistryUrl = schemaRegistry.property("schema.registry.url"),
74+
topic = "events",
75+
schemaJson = schemaJson,
76+
records = listOf(AvroKafkaRecord("key-1", mapOf("id" to 1))),
77+
)
78+
```
79+
4980
## JUnit 5
5081

5182
```kotlin
@@ -148,7 +179,7 @@ GRADLE_USER_HOME=/data/.gradle ./gradlew :example:spring:bootRun --args='--sprin
148179

149180
The JUnit examples in `example/junit` are annotated with `@Disabled`; remove that annotation from an example class to start the configured stack.
150181

151-
The Spark example in `example/spark` creates a `SparkSession` from `BigDataTestKit` endpoints, starts HDFS, Hive Metastore, Kafka, Schema Registry, LocalStack S3, and fake GCS, verifies HDFS and Hive Metastore connectivity, and configures Spark Kafka/S3/GCS settings:
182+
The Spark example in `example/spark` creates a `SparkSession` from `BigDataTestKit` endpoints, starts HDFS, Hive Metastore, Kafka, Schema Registry, LocalStack S3, and fake GCS, then uses `addons` to create the S3 JCEKS file on HDFS and produce Avro records to Kafka. HDFS is used as the config store for the JCEKS file; S3 and GCS are the object stores used by the Iceberg smoke checks.
152183

153184
```bash
154185
GRADLE_USER_HOME=/data/.gradle ./gradlew :example:spark:test

‎addons/build.gradle.kts‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
plugins {
2+
id("buildsrc.convention.kotlin-jvm")
3+
}
4+
5+
description = "Optional bigdata-test helpers for Hadoop credential providers, Kafka, and Avro"
6+
7+
dependencies {
8+
api(libs.hadoopClientApi)
9+
api(libs.hadoopClientRuntime)
10+
api(libs.kafkaAvroSerializer)
11+
api(libs.avro)
12+
13+
implementation(libs.kafkaSchemaRegistryClient)
14+
}

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

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
package org.openprojectx.bigdata.test.addons.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.security.alias.CredentialProviderFactory
7+
import java.net.URI
8+
9+
object HadoopCredentialProviders {
10+
fun hdfsJceksPath(configDir: String, fileName: String = "credentials.jceks"): String =
11+
"jceks://hdfs${configDir.trimEnd('/')}/$fileName"
12+
13+
fun createHdfsJceks(
14+
hdfsUri: String,
15+
configDir: String,
16+
providerPath: String,
17+
credentials: Map<String, String>,
18+
) {
19+
val conf = Configuration(false)
20+
conf.set("fs.defaultFS", hdfsUri)
21+
conf.set(CredentialProviderFactory.CREDENTIAL_PROVIDER_PATH, providerPath)
22+
FileSystem.get(URI.create(hdfsUri), conf).use { fs -> fs.mkdirs(Path(configDir)) }
23+
24+
val provider = CredentialProviderFactory.getProviders(conf).single()
25+
credentials.forEach { (alias, value) ->
26+
if (provider.getCredentialEntry(alias) != null) {
27+
provider.deleteCredentialEntry(alias)
28+
}
29+
provider.createCredentialEntry(alias, value.toCharArray())
30+
}
31+
provider.flush()
32+
}
33+
34+
fun exists(hdfsUri: String, path: String): Boolean {
35+
val conf = Configuration(false)
36+
conf.set("fs.defaultFS", hdfsUri)
37+
return FileSystem.get(URI.create(hdfsUri), conf).use { fs -> fs.exists(Path(path)) }
38+
}
39+
}

‎addons/src/main/kotlin/org/openprojectx/bigdata/test/addons/kafka/KafkaAvroProducers.kt‎

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
package org.openprojectx.bigdata.test.addons.kafka
2+
3+
import io.confluent.kafka.serializers.KafkaAvroSerializer
4+
import org.apache.avro.Schema
5+
import org.apache.avro.generic.GenericData
6+
import org.apache.avro.generic.GenericRecord
7+
import org.apache.kafka.clients.admin.AdminClient
8+
import org.apache.kafka.clients.admin.NewTopic
9+
import org.apache.kafka.clients.producer.KafkaProducer
10+
import org.apache.kafka.clients.producer.ProducerConfig
11+
import org.apache.kafka.clients.producer.ProducerRecord
12+
import org.apache.kafka.common.errors.TopicExistsException
13+
import org.apache.kafka.common.serialization.StringSerializer
14+
import java.util.Properties
15+
16+
data class AvroKafkaRecord(
17+
val key: String,
18+
val values: Map<String, Any?>,
19+
)
20+
21+
object KafkaAvroProducers {
22+
fun produce(
23+
bootstrapServers: String,
24+
schemaRegistryUrl: String,
25+
topic: String,
26+
schemaJson: String,
27+
records: Iterable<AvroKafkaRecord>,
28+
partitions: Int = 1,
29+
replicationFactor: Short = 1,
30+
) {
31+
createTopic(bootstrapServers, topic, partitions, replicationFactor)
32+
33+
val schema = Schema.Parser().parse(schemaJson)
34+
val props = Properties().apply {
35+
put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers)
36+
put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer::class.java.name)
37+
put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer::class.java.name)
38+
put("schema.registry.url", schemaRegistryUrl)
39+
}
40+
41+
KafkaProducer<String, GenericRecord>(props).use { producer ->
42+
records.forEach { record ->
43+
producer.send(ProducerRecord(topic, record.key, record.toGenericRecord(schema))).get()
44+
}
45+
producer.flush()
46+
}
47+
}
48+
49+
fun createTopic(
50+
bootstrapServers: String,
51+
topic: String,
52+
partitions: Int = 1,
53+
replicationFactor: Short = 1,
54+
) {
55+
AdminClient.create(mapOf(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG to bootstrapServers)).use { admin ->
56+
try {
57+
admin.createTopics(listOf(NewTopic(topic, partitions, replicationFactor))).all().get()
58+
} catch (ex: Exception) {
59+
if (ex.cause !is TopicExistsException) throw ex
60+
}
61+
}
62+
}
63+
64+
private fun AvroKafkaRecord.toGenericRecord(schema: Schema): GenericRecord =
65+
GenericData.Record(schema).also { generic ->
66+
values.forEach { (field, value) -> generic.put(field, value) }
67+
}
68+
}

‎example/spark/build.gradle.kts‎

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -21,21 +21,17 @@ configurations.all {
2121

2222
dependencies {
2323
testImplementation(project(":junit5"))
24+
testImplementation(project(":addons"))
2425
testImplementation(libs.sparkSql)
2526
testImplementation(libs.sparkHive)
2627
testImplementation(libs.sparkSqlKafka)
2728
testImplementation(libs.sparkAvro)
2829
testImplementation(libs.hadoopAws)
29-
testImplementation(libs.hadoopClientApi)
30-
testImplementation(libs.hadoopClientRuntime)
3130
testImplementation(libs.icebergSparkRuntime)
3231
testImplementation(libs.icebergAwsBundle)
3332
testImplementation(libs.gcsConnector)
3433
testImplementation(libs.gcsio)
3534
testImplementation(libs.gcsUtilHadoop)
36-
testImplementation(libs.kafkaAvroSerializer)
37-
testImplementation(libs.kafkaSchemaRegistryClient)
38-
testImplementation(libs.avro)
3935
testImplementation(libs.junitJupiterApi)
4036
testRuntimeOnly(libs.junitJupiterEngine)
4137
testRuntimeOnly(libs.junitPlatformLauncher)

‎example/spark/src/test/kotlin/org/openprojectx/bigdata/test/example/spark/SparkBigDataTestExample.kt‎

Lines changed: 23 additions & 67 deletions
Original file line numberDiff line numberDiff line change
@@ -1,22 +1,10 @@
11
package org.openprojectx.bigdata.test.example.spark
22

3-
import io.confluent.kafka.serializers.KafkaAvroSerializer
4-
import org.apache.avro.Schema
5-
import org.apache.avro.generic.GenericData
6-
import org.apache.avro.generic.GenericRecord
7-
import org.apache.hadoop.conf.Configuration
8-
import org.apache.hadoop.fs.FileSystem
9-
import org.apache.hadoop.fs.Path
10-
import org.apache.hadoop.security.alias.CredentialProviderFactory
11-
import org.apache.kafka.clients.admin.AdminClient
12-
import org.apache.kafka.clients.admin.NewTopic
13-
import org.apache.kafka.clients.producer.KafkaProducer
14-
import org.apache.kafka.clients.producer.ProducerConfig
15-
import org.apache.kafka.clients.producer.ProducerRecord
16-
import org.apache.kafka.common.errors.TopicExistsException
17-
import org.apache.kafka.common.serialization.StringSerializer
183
import org.apache.spark.sql.SparkSession
194
import org.junit.jupiter.api.Test
5+
import org.openprojectx.bigdata.test.addons.hadoop.HadoopCredentialProviders
6+
import org.openprojectx.bigdata.test.addons.kafka.AvroKafkaRecord
7+
import org.openprojectx.bigdata.test.addons.kafka.KafkaAvroProducers
208
import org.openprojectx.bigdata.test.core.BigDataService
219
import org.openprojectx.bigdata.test.core.BigDataTestKit
2210
import org.openprojectx.bigdata.test.core.ContainerLogMode
@@ -28,7 +16,6 @@ import java.net.http.HttpResponse
2816
import java.nio.charset.StandardCharsets
2917
import java.nio.file.Files
3018
import java.time.Duration
31-
import java.util.Properties
3219

3320
@BigDataTest(
3421
hdfs = true,
@@ -53,16 +40,18 @@ class SparkBigDataTestExample {
5340
val topic = "spark-avro-events-$runId"
5441
val s3Bucket = "spark-iceberg-s3-$runId"
5542
val gcsBucket = "spark-iceberg-gcs-$runId"
56-
val hdfsConfigDir = Path("/bigdata-test/spark/$runId")
57-
val s3CredentialProviderPath = "jceks://hdfs${hdfsConfigDir}/s3.jceks"
43+
val hdfsConfigDir = "/bigdata-test/spark/$runId"
44+
val s3CredentialProviderPath = HadoopCredentialProviders.hdfsJceksPath(hdfsConfigDir, "s3.jceks")
5845
createS3Bucket(s3.property("aws.endpoint-url.s3"), s3Bucket)
5946
createGcsBucket(gcs.property("google.cloud.storage.host"), gcsBucket)
60-
createS3CredentialProvider(
47+
HadoopCredentialProviders.createHdfsJceks(
6148
hdfsUri = hdfs.property("fs.defaultFS"),
6249
configDir = hdfsConfigDir,
6350
providerPath = s3CredentialProviderPath,
64-
accessKey = s3.property("aws.accessKeyId"),
65-
secretKey = s3.property("aws.secretAccessKey"),
51+
credentials = mapOf(
52+
"fs.s3a.access.key" to s3.property("aws.accessKeyId"),
53+
"fs.s3a.secret.key" to s3.property("aws.secretAccessKey"),
54+
),
6655
)
6756
produceAvroEvents(
6857
bootstrapServers = kafka.property("bootstrap.servers"),
@@ -137,9 +126,10 @@ class SparkBigDataTestExample {
137126
.enableHiveSupport()
138127
.getOrCreate()
139128

140-
private fun assertHdfsConfigStore(spark: SparkSession, hdfsUri: String, configDir: Path) {
141-
FileSystem.get(URI.create(hdfsUri), spark.sparkContext().hadoopConfiguration())
142-
.use { fs -> check(fs.exists(Path(configDir, "s3.jceks"))) }
129+
private fun assertHdfsConfigStore(spark: SparkSession, hdfsUri: String, configDir: String) {
130+
check(HadoopCredentialProviders.exists(hdfsUri, "$configDir/s3.jceks")) {
131+
"Expected S3 JCEKS file in HDFS for ${spark.sparkContext().appName()}"
132+
}
143133
}
144134

145135
private fun assertKafkaAvroInput(spark: SparkSession, bootstrapServers: String, topic: String) {
@@ -179,34 +169,12 @@ class SparkBigDataTestExample {
179169
}
180170

181171

182-
private fun createS3CredentialProvider(
183-
hdfsUri: String,
184-
configDir: Path,
185-
providerPath: String,
186-
accessKey: String,
187-
secretKey: String,
188-
) {
189-
val conf = Configuration(false)
190-
conf.set("fs.defaultFS", hdfsUri)
191-
conf.set(CredentialProviderFactory.CREDENTIAL_PROVIDER_PATH, providerPath)
192-
FileSystem.get(URI.create(hdfsUri), conf).use { fs -> fs.mkdirs(configDir) }
193-
val provider = CredentialProviderFactory.getProviders(conf).single()
194-
provider.createCredentialEntry("fs.s3a.access.key", accessKey.toCharArray())
195-
provider.createCredentialEntry("fs.s3a.secret.key", secretKey.toCharArray())
196-
provider.flush()
197-
}
198-
199172
private fun produceAvroEvents(bootstrapServers: String, schemaRegistryUrl: String, topic: String) {
200-
AdminClient.create(mapOf(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG to bootstrapServers)).use { admin ->
201-
try {
202-
admin.createTopics(listOf(NewTopic(topic, 1, 1))).all().get()
203-
} catch (ex: Exception) {
204-
if (ex.cause !is TopicExistsException) throw ex
205-
}
206-
}
207-
208-
val schema = Schema.Parser().parse(
209-
"""
173+
KafkaAvroProducers.produce(
174+
bootstrapServers = bootstrapServers,
175+
schemaRegistryUrl = schemaRegistryUrl,
176+
topic = topic,
177+
schemaJson = """
210178
{
211179
"type": "record",
212180
"name": "SparkEvent",
@@ -217,23 +185,11 @@ class SparkBigDataTestExample {
217185
]
218186
}
219187
""".trimIndent(),
188+
records = listOf(
189+
AvroKafkaRecord("alpha", mapOf("id" to 1, "name" to "alpha")),
190+
AvroKafkaRecord("beta", mapOf("id" to 2, "name" to "beta")),
191+
),
220192
)
221-
val props = Properties().apply {
222-
put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers)
223-
put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer::class.java.name)
224-
put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer::class.java.name)
225-
put("schema.registry.url", schemaRegistryUrl)
226-
}
227-
KafkaProducer<String, GenericRecord>(props).use { producer ->
228-
listOf("alpha", "beta").forEachIndexed { index, name ->
229-
val record = GenericData.Record(schema).apply {
230-
put("id", index + 1)
231-
put("name", name)
232-
}
233-
producer.send(ProducerRecord(topic, name, record)).get()
234-
}
235-
producer.flush()
236-
}
237193
}
238194

239195
private fun createS3Bucket(endpoint: String, bucket: String) {

0 commit comments

Comments
 (0)