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