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..e6e03dc4 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 @@ -11,12 +11,17 @@ import org.springframework.kafka.listener.ContainerProperties; import org.springframework.kafka.listener.DefaultErrorHandler; import org.springframework.kafka.listener.DeadLetterPublishingRecoverer; +import org.springframework.kafka.listener.DeadLetterPublishingRecoverer.HeaderNames.HeadersToAdd; import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; import org.springframework.lang.NonNull; import org.springframework.util.backoff.FixedBackOff; /** * Kafka-related configuration for {@code cdc-service}. + * + *

Dead-letter records retain the failed record and bounded origin/classification metadata needed + * for authorized recovery, while raw exception messages and stack traces are excluded because they + * can contain database, provider, credential-adjacent, or other deployment-sensitive diagnostics.

*/ @Configuration public class KafkaConfig { @@ -24,6 +29,14 @@ public class KafkaConfig { private static final Logger log = LoggerFactory.getLogger(KafkaConfig.class); private static final int MAX_CONCURRENCY = 32; + /** + * Creates the replica-consumer error handler with bounded retries and a dead-letter fallback. + * + * @param kafkaTemplate Kafka publisher used for dead-letter records + * @param retryBackoffMs delay between retry attempts in milliseconds + * @param retryMaxAttempts maximum retry attempts before dead-letter recovery + * @return configured listener error handler + */ @Bean public DefaultErrorHandler kafkaListenerErrorHandler( @NonNull KafkaTemplate kafkaTemplate, @@ -34,6 +47,7 @@ public DefaultErrorHandler kafkaListenerErrorHandler( kafkaTemplate, (record, ex) -> new TopicPartition(record.topic() + ".DLT", record.partition()) ); + recoverer.excludeHeader(HeadersToAdd.EX_MSG, HeadersToAdd.EX_STACKTRACE); DefaultErrorHandler errorHandler = new DefaultErrorHandler( recoverer, @@ -43,6 +57,14 @@ public DefaultErrorHandler kafkaListenerErrorHandler( return errorHandler; } + /** + * Creates a Kafka listener factory that commits each replica record only after successful handling. + * + * @param consumerFactory Kafka consumer factory + * @param kafkaListenerErrorHandler configured dead-letter/retry handler + * @param concurrency requested listener concurrency + * @return bounded listener-container factory + */ @Bean public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory( @NonNull ConsumerFactory consumerFactory, 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..770dd950 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 @@ -1,7 +1,10 @@ package com.xtrmetl.cdc.config; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.producer.ProducerRecord; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; import org.springframework.classify.BinaryExceptionClassifier; import org.springframework.boot.test.system.CapturedOutput; import org.springframework.boot.test.system.OutputCaptureExtension; @@ -11,15 +14,22 @@ import org.springframework.kafka.listener.ContainerProperties; import org.springframework.kafka.listener.DeadLetterPublishingRecoverer; import org.springframework.kafka.listener.DefaultErrorHandler; +import org.springframework.kafka.support.KafkaHeaders; import org.springframework.test.util.ReflectionTestUtils; import org.springframework.util.backoff.FixedBackOff; +import java.util.concurrent.CompletableFuture; + import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; @SuppressWarnings("unchecked") @ExtendWith(OutputCaptureExtension.class) @@ -53,6 +63,47 @@ void configuresErrorHandlerWithDeadLetterRecovererAndBackOff() { assertFalse(classifier.classify(new IllegalStateException("test"))); } + @Test + void deadLetterRecordKeepsReplayPayloadButDropsRawExceptionMessageAndStackTrace() { + KafkaConfig config = new KafkaConfig(); + KafkaTemplate kafkaTemplate = mock(KafkaTemplate.class); + when(kafkaTemplate.send(any(ProducerRecord.class))) + .thenReturn(CompletableFuture.completedFuture(null)); + + DefaultErrorHandler errorHandler = config.kafkaListenerErrorHandler(kafkaTemplate, 1000L, 30L); + Object failureTracker = ReflectionTestUtils.getField(errorHandler, "failureTracker"); + assertNotNull(failureTracker); + DeadLetterPublishingRecoverer recoverer = + (DeadLetterPublishingRecoverer) ReflectionTestUtils.getField(failureTracker, "recoverer"); + assertNotNull(recoverer); + recoverer.setVerifyPartition(false); + + ConsumerRecord failedRecord = new ConsumerRecord<>( + "xtrmetl-cdc.public.processed_data", + 2, + 41L, + "record-key", + "{\"data\":\"customer-payload\"}" + ); + RuntimeException sensitiveFailure = new RuntimeException( + "jdbc:postgresql://db.internal/prod?user=replica&password=driver-secret" + ); + + recoverer.accept(failedRecord, null, sensitiveFailure); + + ArgumentCaptor> published = ArgumentCaptor.forClass(ProducerRecord.class); + verify(kafkaTemplate).send(published.capture()); + ProducerRecord deadLetterRecord = published.getValue(); + + assertEquals("xtrmetl-cdc.public.processed_data.DLT", deadLetterRecord.topic()); + assertEquals("record-key", deadLetterRecord.key()); + assertEquals("{\"data\":\"customer-payload\"}", deadLetterRecord.value()); + assertNotNull(deadLetterRecord.headers().lastHeader(KafkaHeaders.DLT_ORIGINAL_TOPIC)); + assertNotNull(deadLetterRecord.headers().lastHeader(KafkaHeaders.DLT_EXCEPTION_FQCN)); + assertNull(deadLetterRecord.headers().lastHeader(KafkaHeaders.DLT_EXCEPTION_MESSAGE)); + assertNull(deadLetterRecord.headers().lastHeader(KafkaHeaders.DLT_EXCEPTION_STACKTRACE)); + } + @Test void configuresListenerFactoryWithRecordAckModeAndCommonErrorHandler() { KafkaConfig config = new KafkaConfig();