fix(cdc): keep raw exception diagnostics out of dead-letter headers

cdc-service/src/main/java/com/xtrmetl/cdc/config/KafkaConfig.java

@@ -11,19 +11,32 @@
 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}.
+ *
+ * <p>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.</p>
  */
 @Configuration
 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<String, String> 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<String, String> kafkaListenerContainerFactory(
             @NonNull ConsumerFactory<String, String> consumerFactory,

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<String, String> 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<String, String> 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<ProducerRecord<String, String>> published = ArgumentCaptor.forClass(ProducerRecord.class);
+        verify(kafkaTemplate).send(published.capture());
+        ProducerRecord<String, String> 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();