diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CanonicalChangeRecord.java b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CanonicalChangeRecord.java
index d5fa09d0..e3332e0d 100644
--- a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CanonicalChangeRecord.java
+++ b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CanonicalChangeRecord.java
@@ -1,15 +1,20 @@
package com.xtrmetl.cdc.spi;
import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Objects;
/**
- * Product-neutral change event for any-to-any routing (source → targets).
+ * Immutable product-neutral change event for any-to-any CDC routing.
*
- *
Debezium envelopes are adapted via {@link DebeziumChangeRecordMapper}.
- * Not yet wired into the live Kafka publish path — canonical form is for
- * upcoming multi-target routing.
+ * The constructor takes shallow snapshots of the supplied row maps so later
+ * caller mutations cannot change this record's contents, equality, or hash
+ * identity. Null map references become empty maps, while null database values
+ * inside a supplied map are preserved. Debezium envelopes are adapted via
+ * {@link DebeziumChangeRecordMapper}. The canonical form is not yet wired into
+ * the live Kafka publication path and remains the planned multi-target routing
+ * value object.
*/
public final class CanonicalChangeRecord {
@@ -22,6 +27,18 @@ public final class CanonicalChangeRecord {
private final Map after;
private final Map pk;
+ /**
+ * Creates an immutable canonical CDC record from scalar metadata and row snapshots.
+ *
+ * @param sourceId stable source connector identifier
+ * @param op Debezium-style operation code
+ * @param schema source database schema, when available
+ * @param table source table name, when available
+ * @param tsEpochMs source event timestamp in epoch milliseconds
+ * @param before row values before the change, or {@code null} when unavailable
+ * @param after row values after the change, or {@code null} when unavailable
+ * @param pk primary-key values, or {@code null} when unavailable
+ */
public CanonicalChangeRecord(
String sourceId,
String op,
@@ -37,39 +54,86 @@ public CanonicalChangeRecord(
this.schema = schema;
this.table = table;
this.tsEpochMs = tsEpochMs;
- this.before = before == null ? Map.of() : Collections.unmodifiableMap(before);
- this.after = after == null ? Map.of() : Collections.unmodifiableMap(after);
- this.pk = pk == null ? Map.of() : Collections.unmodifiableMap(pk);
+ this.before = snapshot(before);
+ this.after = snapshot(after);
+ this.pk = snapshot(pk);
}
+ private static Map snapshot(Map source) {
+ if (source == null || source.isEmpty()) {
+ return Map.of();
+ }
+ return Collections.unmodifiableMap(new LinkedHashMap<>(source));
+ }
+
+ /**
+ * Returns the stable source connector identifier.
+ *
+ * @return source connector identifier
+ */
public String getSourceId() {
return sourceId;
}
+ /**
+ * Returns the source operation code.
+ *
+ * @return operation code
+ */
public String getOp() {
return op;
}
+ /**
+ * Returns the source database schema, when available.
+ *
+ * @return source schema
+ */
public String getSchema() {
return schema;
}
+ /**
+ * Returns the source table, when available.
+ *
+ * @return source table
+ */
public String getTable() {
return table;
}
+ /**
+ * Returns the source event timestamp in epoch milliseconds.
+ *
+ * @return source event timestamp
+ */
public long getTsEpochMs() {
return tsEpochMs;
}
+ /**
+ * Returns the immutable construction-time snapshot of values before the change.
+ *
+ * @return immutable before-values map
+ */
public Map getBefore() {
return before;
}
+ /**
+ * Returns the immutable construction-time snapshot of values after the change.
+ *
+ * @return immutable after-values map
+ */
public Map getAfter() {
return after;
}
+ /**
+ * Returns the immutable construction-time snapshot of primary-key values.
+ *
+ * @return immutable primary-key map
+ */
public Map getPk() {
return pk;
}
diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CanonicalChangeRecordTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CanonicalChangeRecordTest.java
new file mode 100644
index 00000000..b2c723b7
--- /dev/null
+++ b/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CanonicalChangeRecordTest.java
@@ -0,0 +1,86 @@
+package com.xtrmetl.cdc.spi;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class CanonicalChangeRecordTest {
+
+ @Test
+ void snapshotsMutableInputMaps() {
+ Map before = new LinkedHashMap<>();
+ before.put("status", "old");
+ Map after = new LinkedHashMap<>();
+ after.put("status", "new");
+ Map pk = new LinkedHashMap<>();
+ pk.put("id", 7L);
+
+ CanonicalChangeRecord record = new CanonicalChangeRecord(
+ "postgres-debezium",
+ "u",
+ "public",
+ "orders",
+ 123L,
+ before,
+ after,
+ pk
+ );
+ int originalHashCode = record.hashCode();
+
+ before.put("status", "mutated");
+ after.clear();
+ pk.put("id", 99L);
+
+ assertEquals("old", record.getBefore().get("status"));
+ assertEquals("new", record.getAfter().get("status"));
+ assertEquals(7L, record.getPk().get("id"));
+ assertEquals(originalHashCode, record.hashCode());
+ }
+
+ @Test
+ void exposesUnmodifiableMaps() {
+ CanonicalChangeRecord record = new CanonicalChangeRecord(
+ "postgres-debezium",
+ "c",
+ "public",
+ "orders",
+ 123L,
+ Map.of(),
+ Map.of("status", "new"),
+ Map.of("id", 7L)
+ );
+
+ assertThrows(UnsupportedOperationException.class,
+ () -> record.getAfter().put("status", "mutated"));
+ assertThrows(UnsupportedOperationException.class,
+ () -> record.getPk().remove("id"));
+ }
+
+ @Test
+ void preservesNullEntriesForDatabaseRows() {
+ Map after = new LinkedHashMap<>();
+ after.put("optional_column", null);
+
+ CanonicalChangeRecord record = new CanonicalChangeRecord(
+ "postgres-debezium",
+ "c",
+ "public",
+ "orders",
+ 123L,
+ null,
+ after,
+ null
+ );
+
+ assertTrue(record.getAfter().containsKey("optional_column"));
+ assertNull(record.getAfter().get("optional_column"));
+ assertTrue(record.getBefore().isEmpty());
+ assertTrue(record.getPk().isEmpty());
+ }
+}