feat(eventbus): add declarative idempotent topic provisioning

libs/fincore-eventbus/src/integrationTest/kotlin/com/fincore/eventbus/EventBusTopicProvisioningIT.kt

@@ -0,0 +1,63 @@
+// SPDX-License-Identifier: BUSL-1.1
+// SPDX-FileCopyrightText: 2026 FinCore Engine Authors
+
+package com.fincore.eventbus
+
+import io.kotest.matchers.shouldBe
+import org.apache.kafka.clients.admin.AdminClient
+import org.junit.jupiter.api.Test
+import org.springframework.boot.autoconfigure.AutoConfigurations
+import org.springframework.boot.test.context.runner.ApplicationContextRunner
+import org.springframework.kafka.core.KafkaAdmin
+import org.testcontainers.junit.jupiter.Container
+import org.testcontainers.junit.jupiter.Testcontainers
+import org.testcontainers.redpanda.RedpandaContainer
+
+@Testcontainers
+class EventBusTopicProvisioningIT {
+    private val runner =
+        ApplicationContextRunner()
+            .withConfiguration(AutoConfigurations.of(EventBusAutoConfiguration::class.java))
+            .withPropertyValues(
+                "fincore.eventbus.bootstrap-servers=${redpanda.bootstrapServers}",
+                "fincore.eventbus.topics[0].name=$TOPIC",
+                "fincore.eventbus.topics[0].partitions=$PARTITIONS",
+                "fincore.eventbus.topics[0].replicas=1",
+            )
+
+    @Test
+    fun `should provision a declared topic with the requested partition count when the context starts`() {
+        runner.run { context ->
+            partitionCount(context.getBean(KafkaAdmin::class.java)) shouldBe PARTITIONS
+        }
+    }
+
+    @Test
+    fun `should re-provision an already existing topic without failing`() {
+        runner.run { context ->
+            partitionCount(context.getBean(KafkaAdmin::class.java)) shouldBe PARTITIONS
+        }
+        // Second context start re-runs provisioning against the same broker and topic: KafkaAdmin
+        // treats the existing same-partition topic as a no-op (create-only, idempotent, INV-1).
+        runner.run { context ->
+            partitionCount(context.getBean(KafkaAdmin::class.java)) shouldBe PARTITIONS
+        }
+    }
+
+    private fun partitionCount(kafkaAdmin: KafkaAdmin): Int =
+        AdminClient.create(kafkaAdmin.configurationProperties).use { client ->
+            val description =
+                client.describeTopics(listOf(TOPIC)).allTopicNames().get()[TOPIC]
+                    ?: error("topic $TOPIC was not provisioned on the broker")
+            description.partitions().size
+        }
+
+    companion object {
+        private const val TOPIC = "fincore.it.provisioned"
+        private const val PARTITIONS = 4
+
+        @Container
+        @JvmStatic
+        val redpanda: RedpandaContainer = RedpandaContainer("redpandadata/redpanda:v24.2.4")
+    }
+}

libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/EventBusAutoConfiguration.kt

@@ -5,6 +5,7 @@ package com.fincore.eventbus
 
 import org.apache.kafka.clients.CommonClientConfigs
 import org.apache.kafka.clients.admin.AdminClientConfig
+import org.apache.kafka.clients.admin.NewTopic
 import org.apache.kafka.clients.producer.ProducerConfig
 import org.apache.kafka.common.config.SaslConfigs
 import org.apache.kafka.common.serialization.StringSerializer
@@ -13,6 +14,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
 import org.springframework.boot.context.properties.EnableConfigurationProperties
 import org.springframework.context.annotation.Bean
+import org.springframework.kafka.config.TopicBuilder
 import org.springframework.kafka.core.DefaultKafkaProducerFactory
 import org.springframework.kafka.core.KafkaAdmin
 import org.springframework.kafka.core.KafkaTemplate
@@ -35,12 +37,26 @@ class EventBusAutoConfiguration {
     @ConditionalOnMissingBean
     fun kafkaAdmin(properties: EventBusProperties): KafkaAdmin = KafkaAdmin(adminConfig(properties))
 
+    @Bean
+    @ConditionalOnMissingBean
+    fun eventBusTopics(properties: EventBusProperties): KafkaAdmin.NewTopics = KafkaAdmin.NewTopics(*newTopics(properties).toTypedArray())
+
     companion object {
         // Durability is a correctness invariant for ledger events, not a deployment knob:
         // acks=all + enable.idempotence=true are fixed so the producer can never be silently
         // weakened into dropping or duplicating events. Connection/security stay configurable.
         private const val DURABLE_ACKS = "all"
 
+        fun newTopics(properties: EventBusProperties): List<NewTopic> =
+            properties.topics.map { spec ->
+                TopicBuilder
+                    .name(spec.name)
+                    .partitions(spec.partitions)
+                    .replicas(spec.replicas)
+                    .configs(spec.configs)
+                    .build()
+            }
+
         fun producerConfig(properties: EventBusProperties): Map<String, Any> =
             buildMap {
                 put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.bootstrapServers)

libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/EventBusProperties.kt

@@ -10,10 +10,29 @@ data class EventBusProperties(
     val bootstrapServers: String,
     val clientId: String = "fincore-eventbus",
     val security: Security = Security(),
+    val topics: List<TopicSpec> = emptyList(),
 ) {
     data class Security(
         val protocol: String = "PLAINTEXT",
         val saslMechanism: String? = null,
         val saslJaasConfig: String? = null,
     )
+
+    data class TopicSpec(
+        val name: String,
+        val partitions: Int = DEFAULT_PARTITIONS,
+        val replicas: Int = DEFAULT_REPLICAS,
+        val configs: Map<String, String> = emptyMap(),
+    ) {
+        init {
+            require(name.isNotBlank()) { "fincore.eventbus.topics[].name must not be blank" }
+            require(partitions > 0) { "fincore.eventbus.topics[].partitions must be positive" }
+            require(replicas > 0) { "fincore.eventbus.topics[].replicas must be positive" }
+        }
+
+        private companion object {
+            const val DEFAULT_PARTITIONS = 3
+            const val DEFAULT_REPLICAS = 1
+        }
+    }
 }

libs/fincore-eventbus/src/test/kotlin/com/fincore/eventbus/EventBusTopicsTest.kt

@@ -0,0 +1,120 @@
+// SPDX-License-Identifier: BUSL-1.1
+// SPDX-FileCopyrightText: 2026 FinCore Engine Authors
+
+package com.fincore.eventbus
+
+import io.kotest.matchers.collections.shouldBeEmpty
+import io.kotest.matchers.collections.shouldHaveSize
+import io.kotest.matchers.nulls.shouldNotBeNull
+import io.kotest.matchers.shouldBe
+import org.junit.jupiter.api.Test
+import org.springframework.boot.autoconfigure.AutoConfigurations
+import org.springframework.boot.test.context.runner.ApplicationContextRunner
+import org.springframework.kafka.core.KafkaAdmin
+
+class EventBusTopicsTest {
+    private val runner =
+        ApplicationContextRunner()
+            .withConfiguration(AutoConfigurations.of(EventBusAutoConfiguration::class.java))
+            .withPropertyValues("fincore.eventbus.bootstrap-servers=localhost:9092")
+
+    @Test
+    fun `should map topic specs to new topics preserving name partitions replicas and configs`() {
+        val properties =
+            EventBusProperties(
+                bootstrapServers = "localhost:9092",
+                topics =
+                    listOf(
+                        EventBusProperties.TopicSpec(
+                            name = "fincore.transaction",
+                            partitions = PARTITIONS,
+                            replicas = REPLICAS,
+                            configs = mapOf("retention.ms" to RETENTION_MS),
+                        ),
+                    ),
+            )
+
+        val topics = EventBusAutoConfiguration.newTopics(properties)
+
+        topics shouldHaveSize 1
+        topics[0].name() shouldBe "fincore.transaction"
+        topics[0].numPartitions() shouldBe PARTITIONS
+        topics[0].replicationFactor() shouldBe REPLICAS.toShort()
+        topics[0].configs()["retention.ms"] shouldBe RETENTION_MS
+    }
+
+    @Test
+    fun `should bind a topic list with a dotted config key when supplied`() {
+        runner
+            .withPropertyValues(
+                "fincore.eventbus.topics[0].name=fincore.transaction",
+                "fincore.eventbus.topics[0].partitions=$PARTITIONS",
+                "fincore.eventbus.topics[0].configs.[retention.ms]=$RETENTION_MS",
+            ).run { context ->
+                val spec = context.getBean(EventBusProperties::class.java).topics.single()
+                spec.name shouldBe "fincore.transaction"
+                spec.partitions shouldBe PARTITIONS
+                spec.configs["retention.ms"] shouldBe RETENTION_MS
+            }
+    }
+
+    @Test
+    fun `should default partitions and replicas when only a name is supplied`() {
+        runner.withPropertyValues("fincore.eventbus.topics[0].name=t").run { context ->
+            val spec = context.getBean(EventBusProperties::class.java).topics.single()
+            spec.partitions shouldBe DEFAULT_PARTITIONS
+            spec.replicas shouldBe DEFAULT_REPLICAS
+        }
+    }
+
+    @Test
+    fun `should reject a blank topic name`() {
+        runner
+            .withPropertyValues(
+                "fincore.eventbus.topics[0].name=",
+                "fincore.eventbus.topics[0].partitions=$PARTITIONS",
+            ).run { context ->
+                context.startupFailure.shouldNotBeNull()
+            }
+    }
+
+    @Test
+    fun `should reject non-positive partitions`() {
+        runner
+            .withPropertyValues(
+                "fincore.eventbus.topics[0].name=t",
+                "fincore.eventbus.topics[0].partitions=0",
+            ).run { context ->
+                context.startupFailure.shouldNotBeNull()
+            }
+    }
+
+    @Test
+    fun `should reject non-positive replicas`() {
+        runner
+            .withPropertyValues(
+                "fincore.eventbus.topics[0].name=t",
+                "fincore.eventbus.topics[0].replicas=0",
+            ).run { context ->
+                context.startupFailure.shouldNotBeNull()
+            }
+    }
+
+    @Test
+    fun `should provision no topics when none are declared`() {
+        runner.run { context ->
+            val properties = context.getBean(EventBusProperties::class.java)
+            properties.topics.shouldBeEmpty()
+            EventBusAutoConfiguration.newTopics(properties).shouldBeEmpty()
+            context.getBean(KafkaAdmin.NewTopics::class.java).shouldNotBeNull()
+        }
+    }
+
+    private companion object {
+        const val PARTITIONS = 6
+        const val REPLICAS = 2
+        const val DEFAULT_PARTITIONS = 3
+        const val DEFAULT_REPLICAS = 1
+        const val RETENTION_MS = "604800000"
+    }
+}