refactor: consolidate addons into extensions for streamlined test setup

- Replaced `addons` module with `extensions` for config-driven integration test setup.
- Migrated `HadoopCredentialProviders` and `KafkaAvroProducers` to `extensions` module.
- Introduced `S3JceksExtension` and `KafkaAvroSeedExtension` for declarative test configurations.
- Updated Spark example to leverage JSON-configured `extensions` for HDFS and Kafka integration.
- Enhanced JUnit support with `BigDataExtensions` annotation for extension configuration.
- Improved modularity and reduced test boilerplate by centralizing setup logic.

Author: CoderYellow

README.md

@@ -6,7 +6,7 @@ Composable Testcontainers-based fixtures for local big-data integration tests.
 
 - `core`: container builder and endpoint/property model
 - `junit5`: `@BigDataTest` extension for JUnit 5 tests
-- `addons`: optional Hadoop, Kafka, and Avro helpers for integration-test setup
+- `extensions`: config-driven extension module for Hadoop credential providers, Kafka, Avro, and future test setup hooks
 - `bigdata-test-spring-boot-autoconfigure`: Spring Boot auto-configuration
 - `bigdata-test-spring-boot-starter`: starter that brings in the auto-configuration
 - `example:spring`: Spring Boot local-development example
@@ -47,36 +47,60 @@ val kit = BigDataTestKit.builder()
     .build()
 ```
 
-## Addons
+## Extensions
 
-`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:
+`core` and `junit5` intentionally stay clean of Hadoop, Spark, and storage-client dependencies. Optional heavier setup lives in `extensions`, which runs config-driven hooks against a started `BigDataTestKit`.
 
-- `HadoopCredentialProviders`: creates HDFS-backed JCEKS files, useful for S3A credentials such as `fs.s3a.access.key` and `fs.s3a.secret.key`.
-- `KafkaAvroProducers`: creates a topic when needed and produces Avro records through Schema Registry.
+The built-in extensions currently support:
+
+- `s3Jceks`: creates an HDFS-backed JCEKS file from the LocalStack S3 endpoint credentials and exposes `s3-jceks.credential-provider.path`.
+- `kafkaAvro`: creates Kafka topics and produces Avro records through Schema Registry from inline JSON records or a records resource.
+
+JUnit usage is declaration-driven. The test declares the services with `@BigDataTest`, then points `@BigDataExtensions` at one or more JSON resources:
 
 ```kotlin
-val configDir = "/bigdata-test/spark/$runId"
-val providerPath = HadoopCredentialProviders.hdfsJceksPath(configDir, "s3.jceks")
-
-HadoopCredentialProviders.createHdfsJceks(
-    hdfsUri = hdfs.property("fs.defaultFS"),
-    configDir = configDir,
-    providerPath = providerPath,
-    credentials = mapOf(
-        "fs.s3a.access.key" to s3.property("aws.accessKeyId"),
-        "fs.s3a.secret.key" to s3.property("aws.secretAccessKey"),
-    ),
+@BigDataExtensions("classpath:bigdata-extensions.json")
+@BigDataTest(
+    hdfs = true,
+    kafka = true,
+    schemaRegistry = true,
+    localStackS3 = true,
 )
+class MyIntegrationTest {
+    @Test
+    fun test(kit: BigDataTestKit, extensions: BigDataExtensionResult) {
+        val providerPath = extensions.required("s3-jceks.credential-provider.path")
+    }
+}
+```
 
-KafkaAvroProducers.produce(
-    bootstrapServers = kafka.property("bootstrap.servers"),
-    schemaRegistryUrl = schemaRegistry.property("schema.registry.url"),
-    topic = "events",
-    schemaJson = schemaJson,
-    records = listOf(AvroKafkaRecord("key-1", mapOf("id" to 1))),
-)
+Example config:
+
+```json
+{
+  "s3Jceks": {
+    "enabled": true,
+    "hdfsDir": "/bigdata-test/spark",
+    "fileName": "s3.jceks"
+  },
+  "kafkaAvro": {
+    "enabled": true,
+    "topics": [
+      {
+        "name": "events",
+        "schema": "classpath:schemas/event.avsc",
+        "records": [
+          {"key": "alpha", "value": {"id": 1, "name": "alpha"}},
+          {"key": "beta", "value": {"id": 2, "name": "beta"}}
+        ]
+      }
+    ]
+  }
+}
 ```
 
+For future extension modules, implement `BigDataExtension` directly for programmatic use or publish a `BigDataExtensionProvider` via `ServiceLoader`. Providers are selected from config entries under `extensions` by `type`, and extensions can hook lifecycle events such as `AFTER_KIT_START`, `BEFORE_TEST_EXECUTION`, and `AFTER_ALL`.
+
 ## JUnit 5
 
 ```kotlin
@@ -179,7 +203,7 @@ GRADLE_USER_HOME=/data/.gradle ./gradlew :example:spring:bootRun --args='--sprin
 
 The JUnit examples in `example/junit` are annotated with `@Disabled`; remove that annotation from an example class to start the configured stack.
 
-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.
+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 `extensions` config 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.
 
 ```bash
 GRADLE_USER_HOME=/data/.gradle ./gradlew :example:spark:test

example/spark/build.gradle.kts

@@ -21,7 +21,7 @@ configurations.all {
 
 dependencies {
     testImplementation(project(":junit5"))
-    testImplementation(project(":addons"))
+    testImplementation(project(":extensions"))
     testImplementation(libs.sparkSql)
     testImplementation(libs.sparkHive)
     testImplementation(libs.sparkSqlKafka)

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

@@ -2,12 +2,11 @@ package org.openprojectx.bigdata.test.example.spark
 
 import org.apache.spark.sql.SparkSession
 import org.junit.jupiter.api.Test
-import org.openprojectx.bigdata.test.addons.hadoop.HadoopCredentialProviders
-import org.openprojectx.bigdata.test.addons.kafka.AvroKafkaRecord
-import org.openprojectx.bigdata.test.addons.kafka.KafkaAvroProducers
 import org.openprojectx.bigdata.test.core.BigDataService
 import org.openprojectx.bigdata.test.core.BigDataTestKit
 import org.openprojectx.bigdata.test.core.ContainerLogMode
+import org.openprojectx.bigdata.test.extensions.core.BigDataExtensionResult
+import org.openprojectx.bigdata.test.extensions.junit5.BigDataExtensions
 import org.openprojectx.bigdata.test.junit5.BigDataTest
 import java.net.URI
 import java.net.http.HttpClient
@@ -17,6 +16,7 @@ import java.nio.charset.StandardCharsets
 import java.nio.file.Files
 import java.time.Duration
 
+@BigDataExtensions("classpath:spark-bigdata-extensions.json")
 @BigDataTest(
     hdfs = true,
     hiveMetastore = true,
@@ -28,7 +28,7 @@ import java.time.Duration
 )
 class SparkBigDataTestExample {
     @Test
-    fun writesAvroKafkaEventsToIcebergOnObjectStorage(kit: BigDataTestKit) {
+    fun writesAvroKafkaEventsToIcebergOnObjectStorage(kit: BigDataTestKit, extensions: BigDataExtensionResult) {
         val hdfs = kit.endpoint(BigDataService.HDFS)
         val hiveMetastore = kit.endpoint(BigDataService.HIVE_METASTORE)
         val kafka = kit.endpoint(BigDataService.KAFKA)
@@ -37,27 +37,13 @@ class SparkBigDataTestExample {
         val gcs = kit.endpoint(BigDataService.FAKE_GCS)
 
         val runId = System.nanoTime().toString()
-        val topic = "spark-avro-events-$runId"
+        val topic = "spark-avro-events"
         val s3Bucket = "spark-iceberg-s3-$runId"
         val gcsBucket = "spark-iceberg-gcs-$runId"
-        val hdfsConfigDir = "/bigdata-test/spark/$runId"
-        val s3CredentialProviderPath = HadoopCredentialProviders.hdfsJceksPath(hdfsConfigDir, "s3.jceks")
+        val s3CredentialProviderPath = extensions.required("s3-jceks.credential-provider.path")
+        val s3CredentialProviderHdfsPath = extensions.required("s3-jceks.hdfs.path")
         createS3Bucket(s3.property("aws.endpoint-url.s3"), s3Bucket)
         createGcsBucket(gcs.property("google.cloud.storage.host"), gcsBucket)
-        HadoopCredentialProviders.createHdfsJceks(
-            hdfsUri = hdfs.property("fs.defaultFS"),
-            configDir = hdfsConfigDir,
-            providerPath = s3CredentialProviderPath,
-            credentials = mapOf(
-                "fs.s3a.access.key" to s3.property("aws.accessKeyId"),
-                "fs.s3a.secret.key" to s3.property("aws.secretAccessKey"),
-            ),
-        )
-        produceAvroEvents(
-            bootstrapServers = kafka.property("bootstrap.servers"),
-            schemaRegistryUrl = schemaRegistry.property("schema.registry.url"),
-            topic = topic,
-        )
 
         sparkSession(
             hdfsUri = hdfs.property("fs.defaultFS"),
@@ -68,7 +54,7 @@ class SparkBigDataTestExample {
             gcsEndpoint = gcs.property("google.cloud.storage.host"),
             gcsBucket = gcsBucket,
         ).use { spark ->
-            assertHdfsConfigStore(spark, hdfs.property("fs.defaultFS"), hdfsConfigDir)
+            assertHdfsConfigStore(spark, hdfs.property("fs.defaultFS"), s3CredentialProviderHdfsPath)
             assertKafkaAvroInput(spark, kafka.property("bootstrap.servers"), topic)
             assertIcebergTable(spark, catalog = "s3", namespace = "demo_$runId", table = "events_s3", storageName = "s3")
             assertIcebergTable(
@@ -126,10 +112,10 @@ class SparkBigDataTestExample {
             .enableHiveSupport()
             .getOrCreate()
 
-    private fun assertHdfsConfigStore(spark: SparkSession, hdfsUri: String, configDir: String) {
-        check(HadoopCredentialProviders.exists(hdfsUri, "$configDir/s3.jceks")) {
-            "Expected S3 JCEKS file in HDFS for ${spark.sparkContext().appName()}"
-        }
+    private fun assertHdfsConfigStore(spark: SparkSession, hdfsUri: String, hdfsPath: String) {
+        val exists = org.apache.hadoop.fs.FileSystem.get(URI.create(hdfsUri), spark.sparkContext().hadoopConfiguration())
+            .use { fs -> fs.exists(org.apache.hadoop.fs.Path(hdfsPath)) }
+        check(exists) { "Expected S3 JCEKS file in HDFS for ${spark.sparkContext().appName()}" }
     }
 
     private fun assertKafkaAvroInput(spark: SparkSession, bootstrapServers: String, topic: String) {
@@ -169,29 +155,6 @@ class SparkBigDataTestExample {
     }
 
 
-    private fun produceAvroEvents(bootstrapServers: String, schemaRegistryUrl: String, topic: String) {
-        KafkaAvroProducers.produce(
-            bootstrapServers = bootstrapServers,
-            schemaRegistryUrl = schemaRegistryUrl,
-            topic = topic,
-            schemaJson = """
-            {
-              "type": "record",
-              "name": "SparkEvent",
-              "namespace": "org.openprojectx.bigdata.test.example.spark",
-              "fields": [
-                {"name": "id", "type": "int"},
-                {"name": "name", "type": "string"}
-              ]
-            }
-            """.trimIndent(),
-            records = listOf(
-                AvroKafkaRecord("alpha", mapOf("id" to 1, "name" to "alpha")),
-                AvroKafkaRecord("beta", mapOf("id" to 2, "name" to "beta")),
-            ),
-        )
-    }
-
     private fun createS3Bucket(endpoint: String, bucket: String) {
         val request = HttpRequest.newBuilder(URI.create("$endpoint/$bucket"))
             .timeout(Duration.ofSeconds(10))

example/spark/src/test/resources/schemas/spark-event.avsc

@@ -0,0 +1,9 @@
+{
+  "type": "record",
+  "name": "SparkEvent",
+  "namespace": "org.openprojectx.bigdata.test.example.spark",
+  "fields": [
+    {"name": "id", "type": "int"},
+    {"name": "name", "type": "string"}
+  ]
+}

example/spark/src/test/resources/spark-bigdata-extensions.json

@@ -0,0 +1,20 @@
+{
+  "s3Jceks": {
+    "enabled": true,
+    "hdfsDir": "/bigdata-test/spark",
+    "fileName": "s3.jceks"
+  },
+  "kafkaAvro": {
+    "enabled": true,
+    "topics": [
+      {
+        "name": "spark-avro-events",
+        "schema": "classpath:schemas/spark-event.avsc",
+        "records": [
+          {"key": "alpha", "value": {"id": 1, "name": "alpha"}},
+          {"key": "beta", "value": {"id": 2, "name": "beta"}}
+        ]
+      }
+    ]
+  }
+}

addons/build.gradle.kts extensions/build.gradle.ktsaddons/build.gradle.kts renamed to extensions/build.gradle.kts

@@ -1,14 +1,17 @@
 plugins {
     id("buildsrc.convention.kotlin-jvm")
+    alias(libs.plugins.kotlinPluginSerialization)
 }
 
-description = "Optional bigdata-test helpers for Hadoop credential providers, Kafka, and Avro"
+description = "Config-driven bigdata-test extensions for Hadoop credential providers, Kafka, and Avro"
 
 dependencies {
+    api(project(":junit5"))
     api(libs.hadoopClientApi)
     api(libs.hadoopClientRuntime)
     api(libs.kafkaAvroSerializer)
     api(libs.avro)
+    api(libs.kotlinxSerialization)
 
     implementation(libs.kafkaSchemaRegistryClient)
 }

extensions/src/main/kotlin/org/openprojectx/bigdata/test/extensions/config/BigDataExtensionsConfigLoader.kt

@@ -0,0 +1,99 @@
+package org.openprojectx.bigdata.test.extensions.config
+
+import kotlinx.serialization.json.Json
+import kotlinx.serialization.json.JsonArray
+import kotlinx.serialization.json.JsonObject
+import kotlinx.serialization.json.JsonPrimitive
+import kotlinx.serialization.json.booleanOrNull
+import kotlinx.serialization.json.contentOrNull
+import kotlinx.serialization.json.intOrNull
+import kotlinx.serialization.json.jsonArray
+import kotlinx.serialization.json.jsonObject
+import kotlinx.serialization.json.jsonPrimitive
+import org.openprojectx.bigdata.test.extensions.core.BigDataExtension
+import org.openprojectx.bigdata.test.extensions.core.BigDataExtensionProvider
+import org.openprojectx.bigdata.test.extensions.core.BigDataExtensionResourceLoader
+import org.openprojectx.bigdata.test.extensions.hadoop.S3JceksExtension
+import org.openprojectx.bigdata.test.extensions.kafka.KafkaAvroRecordSeed
+import org.openprojectx.bigdata.test.extensions.kafka.KafkaAvroSeedExtension
+import org.openprojectx.bigdata.test.extensions.kafka.KafkaAvroTopicSeed
+import java.util.ServiceLoader
+
+class BigDataExtensionsConfigLoader(
+    private val resources: BigDataExtensionResourceLoader = BigDataExtensionResourceLoader(),
+    providers: Iterable<BigDataExtensionProvider> = ServiceLoader.load(BigDataExtensionProvider::class.java),
+) {
+    private val json = Json { ignoreUnknownKeys = true }
+    private val providers = (builtInProviders + providers).associateBy { it.type }
+
+    fun load(locations: Iterable<String>): List<BigDataExtension> =
+        locations.flatMap { load(it) }
+
+    fun load(location: String): List<BigDataExtension> {
+        val root = json.parseToJsonElement(resources.readText(location)).jsonObject
+        val extensions = mutableListOf<BigDataExtension>()
+        root["s3Jceks"]?.jsonObject?.let { config ->
+            if (config.boolean("enabled", default = true)) {
+                extensions += S3JceksExtension(
+                    id = config.string("id", "s3-jceks"),
+                    hdfsDir = config.string("hdfsDir", "/bigdata-test/config"),
+                    fileName = config.string("fileName", "s3.jceks"),
+                    accessKeyAlias = config.string("accessKeyAlias", "fs.s3a.access.key"),
+                    secretKeyAlias = config.string("secretKeyAlias", "fs.s3a.secret.key"),
+                )
+            }
+        }
+        root["kafkaAvro"]?.jsonObject?.let { config ->
+            if (config.boolean("enabled", default = true)) {
+                extensions += KafkaAvroSeedExtension(
+                    id = config.string("id", "kafka-avro-seed"),
+                    topics = config["topics"]?.jsonArray?.map { it.jsonObject.toKafkaAvroTopic() }.orEmpty(),
+                )
+            }
+        }
+        root["extensions"]?.jsonArray?.forEach { item ->
+            val config = item.jsonObject
+            val type = config.string("type")
+            val provider = providers[type] ?: error("Unknown bigdata-test extension type '$type'")
+            extensions += provider.create(config, resources)
+        }
+        return extensions
+    }
+
+    private fun JsonObject.toKafkaAvroTopic(): KafkaAvroTopicSeed {
+        val inlineRecords = this["records"] as? JsonArray
+        val resourceRecords = this["recordsResource"]?.jsonPrimitive?.contentOrNull
+            ?.let { json.parseToJsonElement(resources.readText(it)).jsonArray }
+        val records = (inlineRecords ?: resourceRecords ?: JsonArray(emptyList()))
+            .map { record ->
+                val obj = record.jsonObject
+                KafkaAvroRecordSeed(
+                    key = obj.string("key"),
+                    value = obj["value"]?.jsonObject ?: error("Kafka Avro record requires object field 'value'"),
+                )
+            }
+        return KafkaAvroTopicSeed(
+            name = string("name"),
+            schema = string("schema"),
+            records = records,
+            partitions = int("partitions", 1),
+            replicationFactor = int("replicationFactor", 1).toShort(),
+        )
+    }
+
+    private fun JsonObject.string(name: String): String =
+        this[name]?.jsonPrimitive?.contentOrNull ?: error("Missing required extension config field '$name'")
+
+    private fun JsonObject.string(name: String, default: String): String =
+        this[name]?.jsonPrimitive?.contentOrNull ?: default
+
+    private fun JsonObject.int(name: String, default: Int): Int =
+        this[name]?.jsonPrimitive?.intOrNull ?: default
+
+    private fun JsonObject.boolean(name: String, default: Boolean): Boolean =
+        this[name]?.jsonPrimitive?.booleanOrNull ?: default
+
+    private companion object {
+        val builtInProviders = listOf<BigDataExtensionProvider>()
+    }
+}

extensions/src/main/kotlin/org/openprojectx/bigdata/test/extensions/core/BigDataExtension.kt

@@ -0,0 +1,12 @@
+package org.openprojectx.bigdata.test.extensions.core
+
+import org.openprojectx.bigdata.test.core.BigDataService
+
+interface BigDataExtension {
+    val id: String
+    val requiredServices: Set<BigDataService> get() = emptySet()
+    val events: Set<BigDataExtensionEvent> get() = setOf(BigDataExtensionEvent.AFTER_KIT_START)
+
+    fun onEvent(event: BigDataExtensionEvent, context: BigDataExtensionContext) {
+    }
+}