diff --git a/etl-service/src/main/java/com/xtrmetl/etl/connector/ChangeRecord.java b/etl-service/src/main/java/com/xtrmetl/etl/connector/ChangeRecord.java
index d62bc6c0..c6a51934 100644
--- a/etl-service/src/main/java/com/xtrmetl/etl/connector/ChangeRecord.java
+++ b/etl-service/src/main/java/com/xtrmetl/etl/connector/ChangeRecord.java
@@ -1,15 +1,20 @@
package com.xtrmetl.etl.connector;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import java.util.Objects;
/**
* Normalized change event for target connectors (canonical CDC/ETL record).
*
- *
Row maps are shallow-snapshotted at construction time so callers cannot mutate a
- * record after it has entered a connector pipeline. Null database values remain supported.
+ * JSON-shaped row maps and nested map/list containers are recursively snapshotted at
+ * construction time so later caller mutations cannot change a record that has entered a
+ * connector pipeline. Null database values remain supported. Scalar and other non-container
+ * values are retained as supplied; this type does not claim to clone arbitrary mutable Java
+ * objects.
*/
public final class ChangeRecord {
@@ -101,6 +106,24 @@ private static Map snapshot(Map source) {
if (source == null || source.isEmpty()) {
return Map.of();
}
- return Collections.unmodifiableMap(new LinkedHashMap<>(source));
+ Map copy = new LinkedHashMap<>();
+ source.forEach((key, value) -> copy.put(key, snapshotValue(value)));
+ return Collections.unmodifiableMap(copy);
+ }
+
+ private static Object snapshotValue(Object value) {
+ if (value instanceof Map, ?> mapValue) {
+ Map copy = new LinkedHashMap<>();
+ mapValue.forEach((key, nestedValue) -> copy.put(key, snapshotValue(nestedValue)));
+ return Collections.unmodifiableMap(copy);
+ }
+ if (value instanceof List> listValue) {
+ List copy = new ArrayList<>(listValue.size());
+ for (Object element : listValue) {
+ copy.add(snapshotValue(element));
+ }
+ return Collections.unmodifiableList(copy);
+ }
+ return value;
}
}
diff --git a/etl-service/src/test/java/com/xtrmetl/etl/connector/ChangeRecordTest.java b/etl-service/src/test/java/com/xtrmetl/etl/connector/ChangeRecordTest.java
index 83e814db..f8750fbd 100644
--- a/etl-service/src/test/java/com/xtrmetl/etl/connector/ChangeRecordTest.java
+++ b/etl-service/src/test/java/com/xtrmetl/etl/connector/ChangeRecordTest.java
@@ -2,7 +2,9 @@
import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -41,6 +43,88 @@ void snapshotsMutableInputMaps() {
assertEquals(7L, record.getPk().get("id"));
}
+ @Test
+ void recursivelySnapshotsNestedJsonContainers() {
+ List lineItems = new ArrayList<>();
+ lineItems.add("first");
+ lineItems.add(null);
+ Map details = new LinkedHashMap<>();
+ details.put("line_items", lineItems);
+ Map after = new LinkedHashMap<>();
+ after.put("details", details);
+
+ ChangeRecord record = new ChangeRecord(
+ "source",
+ "u",
+ "public",
+ "orders",
+ 123L,
+ Map.of(),
+ after,
+ Map.of("id", 7L)
+ );
+
+ lineItems.set(0, "mutated");
+ lineItems.add("late");
+ details.put("late_field", "mutated");
+
+ Map, ?> snapshottedDetails = (Map, ?>) record.getAfter().get("details");
+ List> snapshottedLineItems = (List>) snapshottedDetails.get("line_items");
+ assertEquals(2, snapshottedLineItems.size());
+ assertEquals("first", snapshottedLineItems.get(0));
+ assertNull(snapshottedLineItems.get(1));
+ assertTrue(!snapshottedDetails.containsKey("late_field"));
+ }
+
+ @Test
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ void exposesNestedJsonContainersAsUnmodifiable() {
+ List lineItems = new ArrayList<>();
+ lineItems.add("first");
+ Map details = new LinkedHashMap<>();
+ details.put("line_items", lineItems);
+
+ ChangeRecord record = new ChangeRecord(
+ "source",
+ "c",
+ "public",
+ "orders",
+ 123L,
+ Map.of(),
+ Map.of("details", details),
+ Map.of("id", 7L)
+ );
+
+ Map nestedMap = (Map) record.getAfter().get("details");
+ List nestedList = (List) nestedMap.get("line_items");
+ assertThrows(UnsupportedOperationException.class,
+ () -> nestedMap.put("late_field", "mutated"));
+ assertThrows(UnsupportedOperationException.class,
+ () -> nestedList.add("mutated"));
+ }
+
+ @Test
+ void rejectsCyclicNestedContainersDeterministically() {
+ Map cyclic = new LinkedHashMap<>();
+ cyclic.put("self", cyclic);
+
+ IllegalArgumentException failure = assertThrows(
+ IllegalArgumentException.class,
+ () -> new ChangeRecord(
+ "source",
+ "c",
+ "public",
+ "orders",
+ 123L,
+ Map.of(),
+ Map.of("cycle", cyclic),
+ Map.of("id", 7L)
+ )
+ );
+
+ assertEquals("ChangeRecord JSON containers must not contain cycles", failure.getMessage());
+ }
+
@Test
void exposesUnmodifiableMaps() {
ChangeRecord record = new ChangeRecord(