From 770c19c366cdbee0e92faea3312cbea747078cfd Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 10 Aug 2026 17:16:20 +0900 Subject: [PATCH 1/2] test(cdc): require canonical record snapshots --- .../cdc/spi/CanonicalChangeRecordTest.java | 86 +++++++++++++++++++ 1 file changed, 86 insertions(+) create mode 100644 cdc-service/src/test/java/com/xtrmetl/cdc/spi/CanonicalChangeRecordTest.java 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()); + } +} From 621e7c25dae2081534d5fd94f2ac294128d1a8ff Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 10 Aug 2026 17:33:26 +0900 Subject: [PATCH 2/2] fix(cdc): snapshot canonical change record maps --- .../cdc/spi/CanonicalChangeRecord.java | 78 +++++++++++++++++-- 1 file changed, 71 insertions(+), 7 deletions(-) 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; }