diff --git a/libs/fincore-eventbus/src/integrationTest/kotlin/com/fincore/eventbus/EventBusTopicProvisioningIT.kt b/libs/fincore-eventbus/src/integrationTest/kotlin/com/fincore/eventbus/EventBusTopicProvisioningIT.kt new file mode 100644 index 0000000..ac1a690 --- /dev/null +++ b/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") + } +} diff --git a/libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/EventBusAutoConfiguration.kt b/libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/EventBusAutoConfiguration.kt index db2459a..c42953e 100644 --- a/libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/EventBusAutoConfiguration.kt +++ b/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 = + properties.topics.map { spec -> + TopicBuilder + .name(spec.name) + .partitions(spec.partitions) + .replicas(spec.replicas) + .configs(spec.configs) + .build() + } + fun producerConfig(properties: EventBusProperties): Map = buildMap { put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.bootstrapServers) diff --git a/libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/EventBusProperties.kt b/libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/EventBusProperties.kt index 185c55b..94794a3 100644 --- a/libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/EventBusProperties.kt +++ b/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 = 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 = 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 + } + } } diff --git a/libs/fincore-eventbus/src/test/kotlin/com/fincore/eventbus/EventBusTopicsTest.kt b/libs/fincore-eventbus/src/test/kotlin/com/fincore/eventbus/EventBusTopicsTest.kt new file mode 100644 index 0000000..ec37b34 --- /dev/null +++ b/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" + } +}