feat: add S3 and GCS bucket extensions to BigDataExtensions

- Introduced `s3Bucket` and `gcsBucket` extensions for creating buckets in LocalStack S3 and fake GCS server.
- Added bucket identifiers and URI outputs for enhanced configuration.
- Updated `BigDataExtensionsBuilder` with builder methods for declarative bucket setup.
- Enhanced JUnit and programmatic extension support to combine TOML and code-based declarations.
- Updated Spark example, documentation, and user guide to demonstrate the new bucket capabilities.

Author: CoderYellow

README.md

@@ -55,6 +55,7 @@ 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 TOML records or a records resource.
+- `s3Bucket` / `gcsBucket`: creates LocalStack S3 and fake-gcs-server buckets and exposes their bucket names and storage URIs.
 
 JUnit usage is declaration-driven. The test declares the services with `@BigDataTest`, then points `@BigDataExtensions` at one or more TOML resources:
 
@@ -111,6 +112,8 @@ class MyIntegrationTest {
                     record("alpha", mapOf("id" to 1, "name" to "alpha"))
                 }
             }
+            extensions.s3Bucket("spark-iceberg-s3-$suffix", id = "spark-s3-bucket")
+            extensions.gcsBucket("spark-iceberg-gcs-$suffix", id = "spark-gcs-bucket")
         }
     }
 }
@@ -136,6 +139,8 @@ public final class MyExtensionsConfigurer implements BigDataExtensionsConfigurer
 }
 ```
 
+When TOML and programmatic declarations are used together, TOML is loaded first and programmatic declarations are loaded second. Extensions are merged by `extension.id`; if the same id appears more than once, the later declaration replaces the earlier one before any lifecycle hook runs. Use the same id to override file config, or distinct ids to run multiple extensions of the same type.
+
 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

doc/user-guide.adoc

@@ -353,6 +353,9 @@ class MyIntegrationTest {
                     record("beta", mapOf("id" to 2, "name" to "beta"))
                 }
             }
+
+            extensions.s3Bucket("spark-iceberg-s3-$suffix", id = "spark-s3-bucket")
+            extensions.gcsBucket("spark-iceberg-gcs-$suffix", id = "spark-gcs-bucket")
         }
     }
 }
@@ -387,6 +390,25 @@ public final class MyExtensionsConfigurer implements BigDataExtensionsConfigurer
 
 Use `extensions.extension(...)` to add a custom `BigDataExtension` directly when the built-in builder methods are not enough.
 
+When TOML and programmatic declarations are used together, `extension.id` is the identity. If the same id appears more than once, the later declaration replaces the earlier one before any lifecycle hook runs. Programmatic declarations therefore override TOML declarations with the same id. Use distinct ids when multiple extensions of the same type should run.
+
+=== Object Store Bucket Extensions
+
+The S3 and GCS bucket extensions create buckets after the object-store services start. They are useful when bucket names are generated per test class.
+
+[source,kotlin]
+----
+extensions.s3Bucket("spark-iceberg-s3-$suffix", id = "spark-s3-bucket")
+extensions.gcsBucket("spark-iceberg-gcs-$suffix", id = "spark-gcs-bucket")
+----
+
+Outputs:
+
+* `spark-s3-bucket.bucket`
+* `spark-s3-bucket.s3a.uri`
+* `spark-gcs-bucket.bucket`
+* `spark-gcs-bucket.gs.uri`
+
 === S3 JCEKS Extension
 
 The `s3Jceks` extension creates a Hadoop credential provider file in HDFS and stores LocalStack S3 credentials in it.

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

@@ -8,15 +8,12 @@ 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.config.BigDataExtensionsBuilder
+import org.openprojectx.bigdata.test.extensions.config.BigDataExtensionsConfigurer
 import org.openprojectx.bigdata.test.extensions.junit5.BigDataExtensions
 import org.openprojectx.bigdata.test.junit5.BigDataTest
 import java.net.URI
-import java.net.http.HttpClient
-import java.net.http.HttpRequest
-import java.net.http.HttpResponse
-import java.nio.charset.StandardCharsets
 import java.nio.file.Files
-import java.time.Duration
 
 @BigDataExtensions("classpath:spark-bigdata-extensions.toml")
 @BigDataTest(
@@ -29,24 +26,31 @@ import java.time.Duration
     containerLogMode = ContainerLogMode.FILE,
 )
 class SparkBigDataTestExample {
+    companion object : BigDataExtensionsConfigurer {
+        private val runId = System.nanoTime().toString()
+        private const val s3BucketId = "spark-s3-bucket"
+        private const val gcsBucketId = "spark-gcs-bucket"
+
+        override fun configure(extensions: BigDataExtensionsBuilder) {
+            extensions.s3Bucket(bucket = "spark-iceberg-s3-$runId", id = s3BucketId)
+            extensions.gcsBucket(bucket = "spark-iceberg-gcs-$runId", id = gcsBucketId)
+        }
+    }
+
     @Test
     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)
-        val schemaRegistry = kit.endpoint(BigDataService.SCHEMA_REGISTRY)
         val s3 = kit.endpoint(BigDataService.LOCALSTACK_S3)
         val gcs = kit.endpoint(BigDataService.FAKE_GCS)
 
-        val runId = System.nanoTime().toString()
         val topic = "spark-avro-events"
-        val s3Bucket = "spark-iceberg-s3-$runId"
-        val gcsBucket = "spark-iceberg-gcs-$runId"
-        val gcsIcebergDataPath = "gs://$gcsBucket/data/demo_$runId/events_gcs"
+        val s3Bucket = extensions.required("$s3BucketId.bucket")
+        val gcsBucket = extensions.required("$gcsBucketId.bucket")
+        val gcsIcebergDataPath = "${extensions.required("$gcsBucketId.gs.uri")}/data/demo_$runId/events_gcs"
         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)
 
         sparkSession(
             hdfsUri = hdfs.property("fs.defaultFS"),
@@ -266,24 +270,4 @@ class SparkBigDataTestExample {
             client.close()
         }
     }
-
-    private fun createS3Bucket(endpoint: String, bucket: String) {
-        val request = HttpRequest.newBuilder(URI.create("$endpoint/$bucket"))
-            .timeout(Duration.ofSeconds(10))
-            .PUT(HttpRequest.BodyPublishers.noBody())
-            .build()
-        val response = HttpClient.newHttpClient().send(request, HttpResponse.BodyHandlers.discarding())
-        check(response.statusCode() in setOf(200, 409)) { "Failed to create S3 bucket $bucket: HTTP ${response.statusCode()}" }
-    }
-
-    private fun createGcsBucket(endpoint: String, bucket: String) {
-        val request = HttpRequest.newBuilder(URI.create("$endpoint/storage/v1/b?project=bigdata-test"))
-            .timeout(Duration.ofSeconds(10))
-            .header("Content-Type", "application/json")
-            .POST(HttpRequest.BodyPublishers.ofString("""{"name":"$bucket"}""", StandardCharsets.UTF_8))
-            .build()
-        val response = HttpClient.newHttpClient().send(request, HttpResponse.BodyHandlers.discarding())
-        check(response.statusCode() in setOf(200, 201, 409)) { "Failed to create GCS bucket $bucket: HTTP ${response.statusCode()}" }
-    }
-
 }

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

@@ -10,6 +10,8 @@ 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 org.openprojectx.bigdata.test.extensions.objectstore.GcsBucketExtension
+import org.openprojectx.bigdata.test.extensions.objectstore.S3BucketExtension
 import java.util.function.Consumer
 
 class BigDataExtensionsBuilder {
@@ -35,6 +37,26 @@ class BigDataExtensionsBuilder {
         extensions += KafkaAvroBuilder().also { configure.accept(it) }.build()
     }
 
+    fun s3Bucket(bucket: String, id: String = "s3-bucket-$bucket") {
+        extensions += S3BucketExtension(id = id, bucket = bucket)
+    }
+
+    fun s3Bucket(bucket: String) {
+        s3Bucket(bucket = bucket, id = "s3-bucket-$bucket")
+    }
+
+    fun gcsBucket(bucket: String, id: String = "gcs-bucket-$bucket", project: String = "bigdata-test") {
+        extensions += GcsBucketExtension(id = id, bucket = bucket, project = project)
+    }
+
+    fun gcsBucket(bucket: String) {
+        gcsBucket(bucket = bucket, id = "gcs-bucket-$bucket", project = "bigdata-test")
+    }
+
+    fun gcsBucket(bucket: String, id: String) {
+        gcsBucket(bucket = bucket, id = id, project = "bigdata-test")
+    }
+
     internal fun build(): List<BigDataExtension> = extensions.toList()
 }
 

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

@@ -18,6 +18,8 @@ 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 org.openprojectx.bigdata.test.extensions.objectstore.GcsBucketExtension
+import org.openprojectx.bigdata.test.extensions.objectstore.S3BucketExtension
 import java.util.ServiceLoader
 
 class BigDataExtensionsConfigLoader(
@@ -52,6 +54,27 @@ class BigDataExtensionsConfigLoader(
                 )
             }
         }
+        root["s3Buckets"]?.jsonArray?.forEach { item ->
+            val config = item.jsonObject
+            if (config.boolean("enabled", default = true)) {
+                val bucket = config.string("bucket")
+                extensions += S3BucketExtension(
+                    id = config.string("id", "s3-bucket-$bucket"),
+                    bucket = bucket,
+                )
+            }
+        }
+        root["gcsBuckets"]?.jsonArray?.forEach { item ->
+            val config = item.jsonObject
+            if (config.boolean("enabled", default = true)) {
+                val bucket = config.string("bucket")
+                extensions += GcsBucketExtension(
+                    id = config.string("id", "gcs-bucket-$bucket"),
+                    bucket = bucket,
+                    project = config.string("project", "bigdata-test"),
+                )
+            }
+        }
         root["extensions"]?.jsonArray?.forEach { item ->
             val config = item.jsonObject
             val type = config.string("type")

extensions/src/main/kotlin/org/openprojectx/bigdata/test/extensions/junit5/BigDataExtensionsExtension.kt

@@ -9,6 +9,7 @@ import org.openprojectx.bigdata.test.extensions.config.BigDataExtensionsBuilder
 import org.openprojectx.bigdata.test.extensions.config.BigDataExtensionsConfigLoader
 import org.openprojectx.bigdata.test.extensions.config.BigDataExtensionsConfigurer
 import org.openprojectx.bigdata.test.extensions.config.NoopBigDataExtensionsConfigurer
+import org.openprojectx.bigdata.test.extensions.core.BigDataExtension
 import org.openprojectx.bigdata.test.extensions.core.BigDataExtensionEvent
 import org.openprojectx.bigdata.test.extensions.core.BigDataExtensionResult
 import org.openprojectx.bigdata.test.extensions.core.BigDataExtensionRunner
@@ -39,8 +40,10 @@ class BigDataExtensionsExtension : BeforeTestExecutionCallback, AfterAllCallback
         val annotation = context.requiredTestClass.getAnnotation(BigDataExtensions::class.java)
             ?: error("@BigDataExtensions is missing")
         val resources = BigDataExtensionResourceLoader(context.requiredTestClass.classLoader)
-        val extensions = BigDataExtensionsConfigLoader(resources).load(annotation.value.asIterable()) +
-            programmaticExtensions(annotation, context)
+        val extensions = mergeById(
+            BigDataExtensionsConfigLoader(resources).load(annotation.value.asIterable()) +
+                programmaticExtensions(annotation, context),
+        )
         val runner = BigDataExtensionRunner(extensions, resources)
         val kit = BigDataTestKitStore.get(context)
         val result = runner.fire(BigDataExtensionEvent.AFTER_KIT_START, kit)
@@ -53,14 +56,25 @@ class BigDataExtensionsExtension : BeforeTestExecutionCallback, AfterAllCallback
     private fun programmaticExtensions(
         annotation: BigDataExtensions,
         context: ExtensionContext,
-    ): List<org.openprojectx.bigdata.test.extensions.core.BigDataExtension> {
+    ): List<BigDataExtension> {
         val builder = BigDataExtensionsBuilder()
         explicitConfigurer(annotation)?.configure(builder)
         companionConfigurer(context)?.configure(builder)
         instanceConfigurer(context)?.configure(builder)
         return builder.build()
     }
 
+    private fun mergeById(extensions: List<BigDataExtension>): List<BigDataExtension> {
+        val merged = linkedMapOf<String, BigDataExtension>()
+        extensions.forEach { extension ->
+            if (merged.containsKey(extension.id)) {
+                merged.remove(extension.id)
+            }
+            merged[extension.id] = extension
+        }
+        return merged.values.toList()
+    }
+
     private fun explicitConfigurer(annotation: BigDataExtensions): BigDataExtensionsConfigurer? {
         val type = annotation.configurer.java
         if (type == NoopBigDataExtensionsConfigurer::class.java) return null

extensions/src/main/kotlin/org/openprojectx/bigdata/test/extensions/objectstore/ObjectStoreBucketExtensions.kt

@@ -0,0 +1,60 @@
+package org.openprojectx.bigdata.test.extensions.objectstore
+
+import org.openprojectx.bigdata.test.core.BigDataService
+import org.openprojectx.bigdata.test.extensions.core.BigDataExtension
+import org.openprojectx.bigdata.test.extensions.core.BigDataExtensionContext
+import org.openprojectx.bigdata.test.extensions.core.BigDataExtensionEvent
+import java.net.URI
+import java.net.http.HttpClient
+import java.net.http.HttpRequest
+import java.net.http.HttpResponse
+import java.nio.charset.StandardCharsets
+import java.time.Duration
+
+data class S3BucketExtension(
+    override val id: String,
+    val bucket: String,
+) : BigDataExtension {
+    override val requiredServices: Set<BigDataService> = setOf(BigDataService.LOCALSTACK_S3)
+    override val events: Set<BigDataExtensionEvent> = setOf(BigDataExtensionEvent.AFTER_KIT_START)
+
+    override fun onEvent(event: BigDataExtensionEvent, context: BigDataExtensionContext) {
+        val endpoint = context.endpoint(BigDataService.LOCALSTACK_S3).property("aws.endpoint-url.s3")
+        val request = HttpRequest.newBuilder(URI.create("${endpoint.trimEnd('/')}/$bucket"))
+            .timeout(Duration.ofSeconds(10))
+            .PUT(HttpRequest.BodyPublishers.noBody())
+            .build()
+        val response = httpClient.send(request, HttpResponse.BodyHandlers.discarding())
+        check(response.statusCode() in setOf(200, 409)) {
+            "Failed to create S3 bucket $bucket: HTTP ${response.statusCode()}"
+        }
+        context.putOutput("$id.bucket", bucket)
+        context.putOutput("$id.s3a.uri", "s3a://$bucket")
+    }
+}
+
+data class GcsBucketExtension(
+    override val id: String,
+    val bucket: String,
+    val project: String = "bigdata-test",
+) : BigDataExtension {
+    override val requiredServices: Set<BigDataService> = setOf(BigDataService.FAKE_GCS)
+    override val events: Set<BigDataExtensionEvent> = setOf(BigDataExtensionEvent.AFTER_KIT_START)
+
+    override fun onEvent(event: BigDataExtensionEvent, context: BigDataExtensionContext) {
+        val endpoint = context.endpoint(BigDataService.FAKE_GCS).property("google.cloud.storage.host")
+        val request = HttpRequest.newBuilder(URI.create("${endpoint.trimEnd('/')}/storage/v1/b?project=$project"))
+            .timeout(Duration.ofSeconds(10))
+            .header("Content-Type", "application/json")
+            .POST(HttpRequest.BodyPublishers.ofString("""{"name":"$bucket"}""", StandardCharsets.UTF_8))
+            .build()
+        val response = httpClient.send(request, HttpResponse.BodyHandlers.discarding())
+        check(response.statusCode() in setOf(200, 201, 409)) {
+            "Failed to create GCS bucket $bucket: HTTP ${response.statusCode()}"
+        }
+        context.putOutput("$id.bucket", bucket)
+        context.putOutput("$id.gs.uri", "gs://$bucket")
+    }
+}
+
+private val httpClient: HttpClient = HttpClient.newHttpClient()