diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/config/KafkaConfig.java b/cdc-service/src/main/java/com/xtrmetl/cdc/config/KafkaConfig.java
index 5c77a06b..36388b64 100644
--- a/cdc-service/src/main/java/com/xtrmetl/cdc/config/KafkaConfig.java
+++ b/cdc-service/src/main/java/com/xtrmetl/cdc/config/KafkaConfig.java
@@ -23,13 +23,32 @@ public class KafkaConfig {
private static final Logger log = LoggerFactory.getLogger(KafkaConfig.class);
private static final int MAX_CONCURRENCY = 32;
+ private static final String RETRY_BACKOFF_KEY = "xtrmetl.replica.kafka.retry-backoff-ms";
+ private static final String RETRY_MAX_ATTEMPTS_KEY = "xtrmetl.replica.kafka.retry-max-attempts";
+ /**
+ * Builds the replica-listener error handler with bounded retry configuration and terminal
+ * dead-letter recovery.
+ *
+ *
Retry settings are deployment-owned. Negative values are rejected before they reach
+ * Spring's {@link FixedBackOff}, while zero remains a valid explicit choice for immediate
+ * retry or no retry attempts.
+ *
+ * @param kafkaTemplate template used to publish exhausted records to the dead-letter topic
+ * @param retryBackoffMs fixed delay between retry attempts in milliseconds; must be non-negative
+ * @param retryMaxAttempts maximum retry attempts after the original delivery; must be non-negative
+ * @return configured listener error handler
+ * @throws IllegalArgumentException when either retry setting is negative
+ */
@Bean
public DefaultErrorHandler kafkaListenerErrorHandler(
@NonNull KafkaTemplate kafkaTemplate,
@Value("${xtrmetl.replica.kafka.retry-backoff-ms:1000}") long retryBackoffMs,
@Value("${xtrmetl.replica.kafka.retry-max-attempts:30}") long retryMaxAttempts
) {
+ requireNonNegative(RETRY_BACKOFF_KEY, retryBackoffMs);
+ requireNonNegative(RETRY_MAX_ATTEMPTS_KEY, retryMaxAttempts);
+
DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(
kafkaTemplate,
(record, ex) -> new TopicPartition(record.topic() + ".DLT", record.partition())
@@ -77,4 +96,10 @@ public ConcurrentKafkaListenerContainerFactory kafkaListenerCont
factory.setCommonErrorHandler(kafkaListenerErrorHandler);
return factory;
}
+
+ private static void requireNonNegative(String key, long value) {
+ if (value < 0) {
+ throw new IllegalArgumentException(key + " must be greater than or equal to 0");
+ }
+ }
}
diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/config/KafkaConfigTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/config/KafkaConfigTest.java
index 16d9f959..0efd35f8 100644
--- a/cdc-service/src/test/java/com/xtrmetl/cdc/config/KafkaConfigTest.java
+++ b/cdc-service/src/test/java/com/xtrmetl/cdc/config/KafkaConfigTest.java
@@ -18,6 +18,7 @@
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
@@ -53,6 +54,32 @@ void configuresErrorHandlerWithDeadLetterRecovererAndBackOff() {
assertFalse(classifier.classify(new IllegalStateException("test")));
}
+ @Test
+ void rejectsNegativeRetryBackoffBeforeBuildingErrorHandler() {
+ KafkaConfig config = new KafkaConfig();
+ KafkaTemplate kafkaTemplate = mock(KafkaTemplate.class);
+
+ IllegalArgumentException exception = assertThrows(
+ IllegalArgumentException.class,
+ () -> config.kafkaListenerErrorHandler(kafkaTemplate, -1L, 30L)
+ );
+
+ assertTrue(exception.getMessage().contains("xtrmetl.replica.kafka.retry-backoff-ms"));
+ }
+
+ @Test
+ void rejectsNegativeRetryAttemptsBeforeBuildingErrorHandler() {
+ KafkaConfig config = new KafkaConfig();
+ KafkaTemplate kafkaTemplate = mock(KafkaTemplate.class);
+
+ IllegalArgumentException exception = assertThrows(
+ IllegalArgumentException.class,
+ () -> config.kafkaListenerErrorHandler(kafkaTemplate, 1000L, -1L)
+ );
+
+ assertTrue(exception.getMessage().contains("xtrmetl.replica.kafka.retry-max-attempts"));
+ }
+
@Test
void configuresListenerFactoryWithRecordAckModeAndCommonErrorHandler() {
KafkaConfig config = new KafkaConfig();