fix(cdc): prevent dead-letter records from re-entering replica apply

cdc-service/src/main/java/com/xtrmetl/cdc/replication/CdcReplicaConsumer.java

@@ -5,6 +5,12 @@
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.stereotype.Component;
 
+/**
+ * Routes live CDC replica records to the matching replica applier.
+ *
+ * <p>Dead-letter topics are terminal quarantine inputs for operator recovery and are deliberately
+ * ignored by this live replica path, even when a configured topic pattern also matches them.</p>
+ */
 @Component
 @ConditionalOnProperty(prefix = "xtrmetl.replica", name = "enabled", havingValue = "true")
 public class CdcReplicaConsumer {
@@ -13,10 +19,10 @@ public class CdcReplicaConsumer {
     private final SchemaChangeReplicaApplier schemaChangeReplicaApplier;
 
     /**
-     * CDC 복제를 처리하는 소비자 컴포넌트를 생성하고 필요한 복제 적용기(applier)를 주입한다.
+     * Creates the CDC replica consumer with the appliers used for live data and schema events.
      *
-     * @param processedDataReplicaApplier 처리된 데이터 복제 이벤트를 적용하는 객체
-     * @param schemaChangeReplicaApplier 스키마 변경 복제 이벤트를 적용하는 객체
+     * @param processedDataReplicaApplier applies supported data-row events to the replica
+     * @param schemaChangeReplicaApplier applies permitted schema-change events to the replica
      */
     public CdcReplicaConsumer(
             ProcessedDataReplicaApplier processedDataReplicaApplier,
@@ -27,18 +33,23 @@ public CdcReplicaConsumer(
     }
 
     /**
-     * 토픽 접미사에 따라 CDC 복제 메시지를 적절한 레플리카 applier로 라우팅하여 처리한다.
+     * Routes one live CDC record without feeding terminal dead-letter records back into appliers.
      *
-     * <p>레코드의 토픽이 ".schema-changes"로 끝나면 스키마 변경 처리기로 전달하고, 그렇지 않으면 처리된 데이터 복제기로 전달한다.</p>
+     * <p>A topic ending in {@code .DLT} is a terminal dead-letter record and is acknowledged by the
+     * listener without invoking either replica applier. Schema-change topics are routed to the
+     * schema applier; every other live topic keeps the existing data-applier behavior.</p>
      *
-     * @param record Kafka로부터 수신된 레코드(토픽, 키, 값)
+     * @param record Kafka record containing the source topic, key, and Debezium-compatible value
      */
     @KafkaListener(
             topicPattern = "${xtrmetl.replica.topic-pattern}",
             groupId = "${xtrmetl.replica.group-id}"
     )
     public void onMessage(ConsumerRecord<String, String> record) {
         String topic = record.topic();
+        if (topic != null && topic.endsWith(ReplicaTopics.DEAD_LETTER_SUFFIX)) {
+            return;
+        }
         if (topic != null && topic.endsWith(ReplicaTopics.SCHEMA_CHANGES_SUFFIX)) {
             schemaChangeReplicaApplier.apply(topic, record.key(), record.value());
             return;

cdc-service/src/main/java/com/xtrmetl/cdc/replication/ReplicaTopics.java

@@ -3,6 +3,7 @@
 final class ReplicaTopics {
 
     static final String SCHEMA_CHANGES_SUFFIX = ".schema-changes";
+    static final String DEAD_LETTER_SUFFIX = ".DLT";
 
     private ReplicaTopics() {
     }

cdc-service/src/test/java/com/xtrmetl/cdc/replication/CdcReplicaConsumerTest.java

@@ -6,9 +6,10 @@
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.when;
 
 @ExtendWith(MockitoExtension.class)
@@ -42,6 +43,22 @@ void delegatesSchemaChangesTopicToSchemaChangeApplier() {
         verify(processedDataReplicaApplier, never()).apply("xtrmetl-cdc.schema-changes", "k", "v");
     }
 
+    @Test
+    void ignoresDeadLetterTopicSoRecoveryCannotCreateDltChains() {
+        CdcReplicaConsumer consumer = new CdcReplicaConsumer(processedDataReplicaApplier, schemaChangeReplicaApplier);
+        ConsumerRecord<String, String> record = new ConsumerRecord<>(
+                "xtrmetl-cdc.public.processed_data.DLT",
+                0,
+                0L,
+                "k",
+                "poison"
+        );
+
+        consumer.onMessage(record);
+
+        verifyNoInteractions(processedDataReplicaApplier, schemaChangeReplicaApplier);
+    }
+
     @Test
     void delegatesNullTopicToProcessedDataApplier() {
         CdcReplicaConsumer consumer = new CdcReplicaConsumer(processedDataReplicaApplier, schemaChangeReplicaApplier);