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