From 6446571ba002a86d6ae1a8b18c6eee2ed3e43480 Mon Sep 17 00:00:00 2001 From: "@tanya_r" Date: Thu, 18 Jun 2026 02:57:33 -0300 Subject: [PATCH] feat(eventbus): add retry backoff and dead-letter routing with replay Add a consumer error topology: a DefaultErrorHandler with bounded exponential backoff that, once retries are exhausted, routes the record to its dead-letter topic via a DeadLetterPublishingRecoverer, so a poison message can never block a partition. RetryTopicNaming derives the retry and dead-letter topic names from configurable suffixes; the dead-letter route uses partition -1 so the producer places the record by key, preserving per-key affinity. DeadLetterReplayer re-drives a dead-letter topic back to a target topic preserving keys and awaiting each send, so the returned count reflects confirmed re-publishes. All beans are conditional-on-missing and load only when the bus is configured. Covered by unit tests plus a Testcontainers-Redpanda integration test. Closes #193 --- libs/fincore-eventbus/build.gradle.kts | 1 + .../com/fincore/eventbus/retry/RetryDlqIT.kt | 108 ++++++++++++++++++ .../eventbus/EventBusAutoConfiguration.kt | 6 +- .../eventbus/retry/DeadLetterReplayer.kt | 31 +++++ .../eventbus/retry/RetryDlqConfiguration.kt | 63 ++++++++++ .../eventbus/retry/RetryDlqProperties.kt | 38 ++++++ .../eventbus/retry/RetryTopicNaming.kt | 26 +++++ .../eventbus/retry/DeadLetterReplayerTest.kt | 57 +++++++++ .../eventbus/retry/RetryDlqPropertiesTest.kt | 69 +++++++++++ .../eventbus/retry/RetryTopicNamingTest.kt | 28 +++++ 10 files changed, 426 insertions(+), 1 deletion(-) create mode 100644 libs/fincore-eventbus/src/integrationTest/kotlin/com/fincore/eventbus/retry/RetryDlqIT.kt create mode 100644 libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/retry/DeadLetterReplayer.kt create mode 100644 libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/retry/RetryDlqConfiguration.kt create mode 100644 libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/retry/RetryDlqProperties.kt create mode 100644 libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/retry/RetryTopicNaming.kt create mode 100644 libs/fincore-eventbus/src/test/kotlin/com/fincore/eventbus/retry/DeadLetterReplayerTest.kt create mode 100644 libs/fincore-eventbus/src/test/kotlin/com/fincore/eventbus/retry/RetryDlqPropertiesTest.kt create mode 100644 libs/fincore-eventbus/src/test/kotlin/com/fincore/eventbus/retry/RetryTopicNamingTest.kt diff --git a/libs/fincore-eventbus/build.gradle.kts b/libs/fincore-eventbus/build.gradle.kts index 76e6522..421b877 100644 --- a/libs/fincore-eventbus/build.gradle.kts +++ b/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 { diff --git a/libs/fincore-eventbus/src/integrationTest/kotlin/com/fincore/eventbus/retry/RetryDlqIT.kt b/libs/fincore-eventbus/src/integrationTest/kotlin/com/fincore/eventbus/retry/RetryDlqIT.kt new file mode 100644 index 0000000..c4350ff --- /dev/null +++ b/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 + 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 + 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 { + val consumer = KafkaConsumer(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> = + KafkaConsumer(consumerProps("verify")).use { consumer -> + val partition = TopicPartition(topic, 0) + consumer.assign(listOf(partition)) + consumer.seekToBeginning(listOf(partition)) + val collected = mutableListOf>() + 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") + } +} 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 c42953e..722dfb9 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 @@ -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 diff --git a/libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/retry/DeadLetterReplayer.kt b/libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/retry/DeadLetterReplayer.kt new file mode 100644 index 0000000..a3b2411 --- /dev/null +++ b/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, +) { + /** 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, + 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() + } +} diff --git a/libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/retry/RetryDlqConfiguration.kt b/libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/retry/RetryDlqConfiguration.kt new file mode 100644 index 0000000..443f413 --- /dev/null +++ b/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, + 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): DeadLetterReplayer = DeadLetterReplayer(kafkaTemplate) + + private companion object { + const val PARTITION_BY_KEY = -1 + } +} diff --git a/libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/retry/RetryDlqProperties.kt b/libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/retry/RetryDlqProperties.kt new file mode 100644 index 0000000..4d4243a --- /dev/null +++ b/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) + } +} diff --git a/libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/retry/RetryTopicNaming.kt b/libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/retry/RetryTopicNaming.kt new file mode 100644 index 0000000..795371b --- /dev/null +++ b/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" + } +} diff --git a/libs/fincore-eventbus/src/test/kotlin/com/fincore/eventbus/retry/DeadLetterReplayerTest.kt b/libs/fincore-eventbus/src/test/kotlin/com/fincore/eventbus/retry/DeadLetterReplayerTest.kt new file mode 100644 index 0000000..8965b64 --- /dev/null +++ b/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>() + private val kafkaTemplate = mockk>() + 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()) } 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()) } returns records("k1" to "v1") + every { kafkaTemplate.send("payments", "k1", "v1") } returns + CompletableFuture.failedFuture(RuntimeException("broker down")) + + shouldThrow { + replayer.replay(consumer, "payments", Duration.ofSeconds(1)) + } + } + + private fun records(vararg entries: Pair): ConsumerRecords { + 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> = CompletableFuture.completedFuture(mockk(relaxed = true)) +} diff --git a/libs/fincore-eventbus/src/test/kotlin/com/fincore/eventbus/retry/RetryDlqPropertiesTest.kt b/libs/fincore-eventbus/src/test/kotlin/com/fincore/eventbus/retry/RetryDlqPropertiesTest.kt new file mode 100644 index 0000000..28c7267 --- /dev/null +++ b/libs/fincore-eventbus/src/test/kotlin/com/fincore/eventbus/retry/RetryDlqPropertiesTest.kt @@ -0,0 +1,69 @@ +// 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.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.listener.DefaultErrorHandler +import java.time.Duration + +class RetryDlqPropertiesTest { + private val runner = + ApplicationContextRunner() + .withConfiguration(AutoConfigurations.of(EventBusAutoConfiguration::class.java)) + .withPropertyValues("fincore.eventbus.bootstrap-servers=localhost:9092") + + @Test + fun `should expose safe defaults when no overrides are supplied`() { + runner.run { context -> + val props = context.getBean(RetryDlqProperties::class.java) + props.maxAttempts shouldBe 3 + props.backoffMultiplier shouldBe 2.0 + props.retrySuffix shouldBe "-retry" + props.deadLetterSuffix shouldBe "-dlt" + } + } + + @Test + fun `should bind explicit overrides when supplied`() { + runner + .withPropertyValues( + "fincore.eventbus.retry.max-attempts=5", + "fincore.eventbus.retry.initial-backoff=250ms", + "fincore.eventbus.retry.backoff-multiplier=3.0", + "fincore.eventbus.retry.dead-letter-suffix=.DLT", + ).run { context -> + val props = context.getBean(RetryDlqProperties::class.java) + props.maxAttempts shouldBe 5 + props.initialBackoff shouldBe Duration.ofMillis(250) + props.backoffMultiplier shouldBe 3.0 + props.deadLetterSuffix shouldBe ".DLT" + } + } + + @Test + fun `should reject a non-positive max attempts`() { + runner.withPropertyValues("fincore.eventbus.retry.max-attempts=0").run { context -> + context.startupFailure.shouldNotBeNull() + } + } + + @Test + fun `should reject a backoff multiplier below one`() { + runner.withPropertyValues("fincore.eventbus.retry.backoff-multiplier=0.5").run { context -> + context.startupFailure.shouldNotBeNull() + } + } + + @Test + fun `should register a default error handler when a kafka template is present`() { + runner.run { context -> + context.getBean(DefaultErrorHandler::class.java).shouldNotBeNull() + } + } +} diff --git a/libs/fincore-eventbus/src/test/kotlin/com/fincore/eventbus/retry/RetryTopicNamingTest.kt b/libs/fincore-eventbus/src/test/kotlin/com/fincore/eventbus/retry/RetryTopicNamingTest.kt new file mode 100644 index 0000000..8e40a08 --- /dev/null +++ b/libs/fincore-eventbus/src/test/kotlin/com/fincore/eventbus/retry/RetryTopicNamingTest.kt @@ -0,0 +1,28 @@ +// 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 org.junit.jupiter.api.Test + +class RetryTopicNamingTest { + private val naming = RetryTopicNaming(retrySuffix = "-retry", deadLetterSuffix = "-dlt") + + @Test + fun `should append the dead-letter suffix to the base topic`() { + naming.deadLetterTopic("fincore.transaction") shouldBe "fincore.transaction-dlt" + } + + @Test + fun `should build a tiered retry topic name from the base topic and tier`() { + naming.retryTopic("fincore.transaction", 1) shouldBe "fincore.transaction-retry-1" + } + + @Test + fun `should reject a blank base topic`() { + shouldThrow { naming.deadLetterTopic(" ") } + shouldThrow { naming.retryTopic(" ", 0) } + } +}