feat(eventbus): add retry backoff and dead-letter routing with replay

libs/fincore-eventbus/build.gradle.kts

@@ -34,6 +34,7 @@ dependencies {
     testImplementation(libs.spring.boot.starter.test)
     testImplementation(libs.kotest.assertions.core)
     testImplementation(libs.kotest.runner.junit5)
+    testImplementation(libs.mockk)
 }
 
 tasks.test {

libs/fincore-eventbus/src/integrationTest/kotlin/com/fincore/eventbus/retry/RetryDlqIT.kt

@@ -0,0 +1,108 @@
+// SPDX-License-Identifier: BUSL-1.1
+// SPDX-FileCopyrightText: 2026 FinCore Engine Authors
+
+package com.fincore.eventbus.retry
+
+import com.fincore.eventbus.EventBusAutoConfiguration
+import io.kotest.matchers.shouldBe
+import org.apache.kafka.clients.consumer.ConsumerConfig
+import org.apache.kafka.clients.consumer.ConsumerRecord
+import org.apache.kafka.clients.consumer.KafkaConsumer
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.serialization.StringDeserializer
+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.KafkaTemplate
+import org.springframework.kafka.listener.DeadLetterPublishingRecoverer
+import org.testcontainers.junit.jupiter.Container
+import org.testcontainers.junit.jupiter.Testcontainers
+import org.testcontainers.redpanda.RedpandaContainer
+import java.time.Duration
+import java.time.Instant
+import java.util.Properties
+
+@Testcontainers
+class RetryDlqIT {
+    private val runner =
+        ApplicationContextRunner()
+            .withConfiguration(AutoConfigurations.of(EventBusAutoConfiguration::class.java))
+            .withPropertyValues("fincore.eventbus.bootstrap-servers=${redpanda.bootstrapServers}")
+
+    @Test
+    fun `should route a recovered record to its dead-letter topic`() {
+        runner.run { context ->
+            @Suppress("UNCHECKED_CAST")
+            val template = context.getBean(KafkaTemplate::class.java) as KafkaTemplate<String, String>
+            val recoverer = context.getBean(DeadLetterPublishingRecoverer::class.java)
+            val record = ConsumerRecord("orders", 0, 0L, "acc_1", "payload-1")
+
+            recoverer.accept(record, RuntimeException("boom"))
+            template.flush()
+
+            val dead = consume("orders-dlt", expected = 1)
+            dead.single().key() shouldBe "acc_1"
+            dead.single().value() shouldBe "payload-1"
+        }
+    }
+
+    @Test
+    fun `should replay dead-lettered records back to the target topic preserving keys`() {
+        runner.run { context ->
+            @Suppress("UNCHECKED_CAST")
+            val template = context.getBean(KafkaTemplate::class.java) as KafkaTemplate<String, String>
+            val replayer = context.getBean(DeadLetterReplayer::class.java)
+            template.send("payments-dlt", "p_1", "amount-1").get()
+            template.send("payments-dlt", "p_2", "amount-2").get()
+            template.flush()
+
+            val replayed =
+                dltConsumer("payments-dlt").use { consumer ->
+                    replayer.replay(consumer, "payments", Duration.ofSeconds(POLL_TIMEOUT_SECONDS))
+                }
+
+            replayed shouldBe 2
+            consume("payments", expected = 2).map { it.key() }.toSet() shouldBe setOf("p_1", "p_2")
+        }
+    }
+
+    private fun dltConsumer(topic: String): KafkaConsumer<String, String> {
+        val consumer = KafkaConsumer<String, String>(consumerProps("replay"))
+        val partition = TopicPartition(topic, 0)
+        consumer.assign(listOf(partition))
+        consumer.seekToBeginning(listOf(partition))
+        return consumer
+    }
+
+    private fun consume(
+        topic: String,
+        expected: Int,
+    ): List<ConsumerRecord<String, String>> =
+        KafkaConsumer<String, String>(consumerProps("verify")).use { consumer ->
+            val partition = TopicPartition(topic, 0)
+            consumer.assign(listOf(partition))
+            consumer.seekToBeginning(listOf(partition))
+            val collected = mutableListOf<ConsumerRecord<String, String>>()
+            val deadline = Instant.now().plusSeconds(POLL_TIMEOUT_SECONDS)
+            while (collected.size < expected && Instant.now().isBefore(deadline)) {
+                consumer.poll(Duration.ofSeconds(1)).forEach { collected.add(it) }
+            }
+            collected
+        }
+
+    private fun consumerProps(group: String): Properties =
+        Properties().apply {
+            this[ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG] = redpanda.bootstrapServers
+            this[ConsumerConfig.GROUP_ID_CONFIG] = "retry-dlq-it-$group"
+            this[ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG] = StringDeserializer::class.java
+            this[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = StringDeserializer::class.java
+        }
+
+    companion object {
+        private const val POLL_TIMEOUT_SECONDS = 15L
+
+        @Container
+        @JvmStatic
+        val redpanda: RedpandaContainer = RedpandaContainer("redpandadata/redpanda:v24.2.4")
+    }
+}

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

@@ -3,6 +3,8 @@
 
 package com.fincore.eventbus
 
+import com.fincore.eventbus.retry.RetryDlqConfiguration
+import com.fincore.eventbus.retry.RetryDlqProperties
 import org.apache.kafka.clients.CommonClientConfigs
 import org.apache.kafka.clients.admin.AdminClientConfig
 import org.apache.kafka.clients.admin.NewTopic
@@ -14,6 +16,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.context.annotation.Import
 import org.springframework.kafka.config.TopicBuilder
 import org.springframework.kafka.core.DefaultKafkaProducerFactory
 import org.springframework.kafka.core.KafkaAdmin
@@ -22,7 +25,8 @@ import org.springframework.kafka.core.ProducerFactory
 
 @AutoConfiguration
 @ConditionalOnProperty(prefix = "fincore.eventbus", name = ["bootstrap-servers"])
-@EnableConfigurationProperties(EventBusProperties::class)
+@EnableConfigurationProperties(EventBusProperties::class, RetryDlqProperties::class)
+@Import(RetryDlqConfiguration::class)
 class EventBusAutoConfiguration {
     @Bean
     @ConditionalOnMissingBean

libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/retry/DeadLetterReplayer.kt

@@ -0,0 +1,31 @@
+// SPDX-License-Identifier: BUSL-1.1
+// SPDX-FileCopyrightText: 2026 FinCore Engine Authors
+
+package com.fincore.eventbus.retry
+
+import org.apache.kafka.clients.consumer.Consumer
+import org.springframework.kafka.core.KafkaTemplate
+import java.time.Duration
+
+/**
+ * Re-drives dead-lettered records back to a target topic. Polls a caller-supplied consumer (already
+ * subscribed to the dead-letter topic, so offset and commit policy stay with the caller) once and
+ * re-publishes each record preserving its key. Non-destructive: the dead-letter topic is not mutated.
+ */
+class DeadLetterReplayer(
+    private val kafkaTemplate: KafkaTemplate<String, String>,
+) {
+    /** Returns the number of records confirmed re-published. Throws if any send fails, so an
+     *  incident-recovery caller never sees a success count for records that did not land. */
+    fun replay(
+        consumer: Consumer<String, String>,
+        targetTopic: String,
+        pollTimeout: Duration,
+    ): Int {
+        val records = consumer.poll(pollTimeout)
+        records
+            .map { record -> kafkaTemplate.send(targetTopic, record.key(), record.value()) }
+            .forEach { it.get() }
+        return records.count()
+    }
+}

libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/retry/RetryDlqConfiguration.kt

@@ -0,0 +1,63 @@
+// SPDX-License-Identifier: BUSL-1.1
+// SPDX-FileCopyrightText: 2026 FinCore Engine Authors
+
+package com.fincore.eventbus.retry
+
+import org.apache.kafka.common.TopicPartition
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.kafka.core.KafkaTemplate
+import org.springframework.kafka.listener.DeadLetterPublishingRecoverer
+import org.springframework.kafka.listener.DefaultErrorHandler
+import org.springframework.util.backoff.ExponentialBackOff
+
+/**
+ * Consumer error topology: bounded exponential-backoff retry, then terminal routing of an exhausted
+ * record to its dead-letter topic. A consumer attaches [kafkaErrorHandler] to its listener container
+ * factory. Loaded only when the event bus is configured (imported by EventBusAutoConfiguration).
+ */
+@Configuration
+class RetryDlqConfiguration {
+    @Bean
+    @ConditionalOnMissingBean
+    fun retryTopicNaming(properties: RetryDlqProperties): RetryTopicNaming =
+        RetryTopicNaming(properties.retrySuffix, properties.deadLetterSuffix)
+
+    @Bean
+    @ConditionalOnMissingBean
+    fun deadLetterPublishingRecoverer(
+        kafkaTemplate: KafkaTemplate<String, String>,
+        naming: RetryTopicNaming,
+    ): DeadLetterPublishingRecoverer =
+        DeadLetterPublishingRecoverer(kafkaTemplate) { record, _ ->
+            // partition -1 lets the producer place the record by key, preserving per-key affinity
+            // without assuming the dead-letter topic has the same partition count as the source.
+            TopicPartition(naming.deadLetterTopic(record.topic()), PARTITION_BY_KEY)
+        }
+
+    @Bean
+    @ConditionalOnMissingBean
+    fun kafkaErrorHandler(
+        recoverer: DeadLetterPublishingRecoverer,
+        properties: RetryDlqProperties,
+    ): DefaultErrorHandler =
+        DefaultErrorHandler(
+            recoverer,
+            ExponentialBackOff().apply {
+                initialInterval = properties.initialBackoff.toMillis()
+                multiplier = properties.backoffMultiplier
+                maxInterval = properties.maxBackoff.toMillis()
+                // the initial delivery counts as attempt 1, so this caps the number of RETRIES
+                maxAttempts = properties.maxAttempts - 1
+            },
+        )
+
+    @Bean
+    @ConditionalOnMissingBean
+    fun deadLetterReplayer(kafkaTemplate: KafkaTemplate<String, String>): DeadLetterReplayer = DeadLetterReplayer(kafkaTemplate)
+
+    private companion object {
+        const val PARTITION_BY_KEY = -1
+    }
+}

libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/retry/RetryDlqProperties.kt

@@ -0,0 +1,38 @@
+// SPDX-License-Identifier: BUSL-1.1
+// SPDX-FileCopyrightText: 2026 FinCore Engine Authors
+
+package com.fincore.eventbus.retry
+
+import org.springframework.boot.context.properties.ConfigurationProperties
+import org.springframework.validation.annotation.Validated
+import java.time.Duration
+
+@Validated
+@ConfigurationProperties(prefix = "fincore.eventbus.retry")
+data class RetryDlqProperties(
+    val maxAttempts: Int = DEFAULT_MAX_ATTEMPTS,
+    val initialBackoff: Duration = defaultInitialBackoff,
+    val backoffMultiplier: Double = DEFAULT_MULTIPLIER,
+    val maxBackoff: Duration = defaultMaxBackoff,
+    val retrySuffix: String = "-retry",
+    val deadLetterSuffix: String = "-dlt",
+) {
+    init {
+        require(maxAttempts >= 1) { "fincore.eventbus.retry.max-attempts must be at least 1" }
+        require(backoffMultiplier >= MIN_MULTIPLIER) { "fincore.eventbus.retry.backoff-multiplier must be >= 1.0" }
+        require(!initialBackoff.isNegative && !initialBackoff.isZero) {
+            "fincore.eventbus.retry.initial-backoff must be positive"
+        }
+        require(!maxBackoff.isNegative && !maxBackoff.isZero) { "fincore.eventbus.retry.max-backoff must be positive" }
+        require(retrySuffix.isNotBlank()) { "fincore.eventbus.retry.retry-suffix must not be blank" }
+        require(deadLetterSuffix.isNotBlank()) { "fincore.eventbus.retry.dead-letter-suffix must not be blank" }
+    }
+
+    private companion object {
+        const val DEFAULT_MAX_ATTEMPTS = 3
+        const val DEFAULT_MULTIPLIER = 2.0
+        const val MIN_MULTIPLIER = 1.0
+        val defaultInitialBackoff: Duration = Duration.ofSeconds(1)
+        val defaultMaxBackoff: Duration = Duration.ofSeconds(10)
+    }
+}

libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/retry/RetryTopicNaming.kt

@@ -0,0 +1,26 @@
+// SPDX-License-Identifier: BUSL-1.1
+// SPDX-FileCopyrightText: 2026 FinCore Engine Authors
+
+package com.fincore.eventbus.retry
+
+/**
+ * Suffix-based naming for the retry and dead-letter topics of a base topic. Generic - the caller
+ * supplies the base topic; no business names are encoded here.
+ */
+class RetryTopicNaming(
+    private val retrySuffix: String,
+    private val deadLetterSuffix: String,
+) {
+    fun deadLetterTopic(baseTopic: String): String {
+        require(baseTopic.isNotBlank()) { "baseTopic must not be blank" }
+        return "$baseTopic$deadLetterSuffix"
+    }
+
+    fun retryTopic(
+        baseTopic: String,
+        tier: Int,
+    ): String {
+        require(baseTopic.isNotBlank()) { "baseTopic must not be blank" }
+        return "$baseTopic$retrySuffix-$tier"
+    }
+}

libs/fincore-eventbus/src/test/kotlin/com/fincore/eventbus/retry/DeadLetterReplayerTest.kt

@@ -0,0 +1,57 @@
+// SPDX-License-Identifier: BUSL-1.1
+// SPDX-FileCopyrightText: 2026 FinCore Engine Authors
+
+package com.fincore.eventbus.retry
+
+import io.kotest.assertions.throwables.shouldThrow
+import io.kotest.matchers.shouldBe
+import io.mockk.every
+import io.mockk.mockk
+import io.mockk.verify
+import org.apache.kafka.clients.consumer.Consumer
+import org.apache.kafka.clients.consumer.ConsumerRecord
+import org.apache.kafka.clients.consumer.ConsumerRecords
+import org.apache.kafka.common.TopicPartition
+import org.junit.jupiter.api.Test
+import org.springframework.kafka.core.KafkaTemplate
+import org.springframework.kafka.support.SendResult
+import java.time.Duration
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.ExecutionException
+
+class DeadLetterReplayerTest {
+    private val consumer = mockk<Consumer<String, String>>()
+    private val kafkaTemplate = mockk<KafkaTemplate<String, String>>()
+    private val replayer = DeadLetterReplayer(kafkaTemplate)
+
+    @Test
+    fun `should republish each polled record to the target topic and return the confirmed count`() {
+        every { consumer.poll(any<Duration>()) } returns records("k1" to "v1", "k2" to "v2")
+        every { kafkaTemplate.send("payments", any(), any()) } returns completed()
+
+        val replayed = replayer.replay(consumer, "payments", Duration.ofSeconds(1))
+
+        replayed shouldBe 2
+        verify { kafkaTemplate.send("payments", "k1", "v1") }
+        verify { kafkaTemplate.send("payments", "k2", "v2") }
+    }
+
+    @Test
+    fun `should propagate the failure when a republish does not confirm`() {
+        every { consumer.poll(any<Duration>()) } returns records("k1" to "v1")
+        every { kafkaTemplate.send("payments", "k1", "v1") } returns
+            CompletableFuture.failedFuture(RuntimeException("broker down"))
+
+        shouldThrow<ExecutionException> {
+            replayer.replay(consumer, "payments", Duration.ofSeconds(1))
+        }
+    }
+
+    private fun records(vararg entries: Pair<String, String>): ConsumerRecords<String, String> {
+        val partition = TopicPartition("payments-dlt", 0)
+        val list = entries.mapIndexed { i, (k, v) -> ConsumerRecord("payments-dlt", 0, i.toLong(), k, v) }
+        return ConsumerRecords(mapOf(partition to list))
+    }
+
+    private fun completed(): CompletableFuture<SendResult<String, String>> = CompletableFuture.completedFuture(mockk(relaxed = true))
+}