repair(cdc): replay Kafka retry validation on live develop

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.
+     *
+     * <p>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.</p>
+     *
+     * @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<String, String> 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<String, String> 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");
+        }
+    }
 }

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<String, String> 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<String, String> 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();