From 70a748a967fd6ad8e10e2d19c9ca7d3ce996da6d Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 1 Sep 2026 21:47:24 +0900 Subject: [PATCH 01/36] test(cdc): require semantic source identifiers --- .../CdcSourceFactoryIdentityContractTest.java | 26 +++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcSourceFactoryIdentityContractTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcSourceFactoryIdentityContractTest.java index 6bd004b7..a1082e36 100644 --- a/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcSourceFactoryIdentityContractTest.java +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcSourceFactoryIdentityContractTest.java @@ -3,8 +3,11 @@ import org.junit.jupiter.api.Test; import java.util.List; +import java.util.Map; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -13,6 +16,29 @@ */ class CdcSourceFactoryIdentityContractTest { + @Test + void sourceSpecPublishesSemanticIdentifiers() { + CdcSourceFactory.SourceSpec sourceSpec = + new CdcSourceFactory.SourceSpec("pg-main", "postgres-debezium", true); + + assertEquals("pg-main", sourceSpec.sourceId()); + assertEquals("postgres-debezium", sourceSpec.sourceType()); + } + + @Test + void configuredSourceDescriptionKeepsLegacyWireKeysAtCompatibilityBoundary() { + CdcSourceFactory factory = factory(); + + Map sourceDescription = factory.describeConfigured(List.of( + new CdcSourceFactory.SourceSpec("pg-main", "postgres-debezium", true) + )).getFirst(); + + assertEquals("pg-main", sourceDescription.get("id")); + assertEquals("postgres-debezium", sourceDescription.get("type")); + assertFalse(sourceDescription.containsKey("sourceId")); + assertFalse(sourceDescription.containsKey("sourceType")); + } + @Test void duplicateConfiguredSourceIdsFailClosed() { CdcSourceFactory factory = factory(); From f48f800a545fff96a9b9aa7eecae53ae35585110 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 1 Sep 2026 21:49:03 +0900 Subject: [PATCH 02/36] refactor(cdc): make source spec identifiers explicit --- .../com/xtrmetl/cdc/spi/CdcSourceFactory.java | 62 ++++++++++--------- 1 file changed, 33 insertions(+), 29 deletions(-) diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcSourceFactory.java b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcSourceFactory.java index d9be8af1..e09918e2 100644 --- a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcSourceFactory.java +++ b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcSourceFactory.java @@ -19,62 +19,66 @@ @Component public class CdcSourceFactory { - private final CdcSourceRegistry registry; + private final CdcSourceRegistry sourceRegistry; - public CdcSourceFactory(CdcSourceRegistry registry) { - this.registry = registry; + public CdcSourceFactory(CdcSourceRegistry sourceRegistry) { + this.sourceRegistry = sourceRegistry; } - public Optional resolve(String type) { - if (type == null || type.isBlank()) { + public Optional resolve(String sourceType) { + if (sourceType == null || sourceType.isBlank()) { return Optional.empty(); } - String normalized = type.trim().toLowerCase(Locale.ROOT); - return registry.find(normalized); + String normalizedSourceType = sourceType.trim().toLowerCase(Locale.ROOT); + return sourceRegistry.find(normalizedSourceType); } /** * Validates and describes a multi-source configuration list without starting engines. * Live capture remains single-source in {@code CdcService}. * - * @param specs configured source entries; {@code null} is treated as an empty list + * @param sourceSpecs configured source entries; {@code null} is treated as an empty list * @return one descriptive row for each configured source * @throws IllegalArgumentException when two entries declare the same source id */ - public List> describeConfigured(List specs) { - List> out = new ArrayList<>(); - if (specs == null) { - return out; + public List> describeConfigured(List sourceSpecs) { + List> sourceDescriptions = new ArrayList<>(); + if (sourceSpecs == null) { + return sourceDescriptions; } Set sourceIds = new HashSet<>(); - for (SourceSpec spec : specs) { - if (!sourceIds.add(spec.id())) { - throw new IllegalArgumentException("duplicate source id: " + spec.id()); + for (SourceSpec sourceSpec : sourceSpecs) { + if (!sourceIds.add(sourceSpec.sourceId())) { + throw new IllegalArgumentException("duplicate source id: " + sourceSpec.sourceId()); } - Map row = new java.util.LinkedHashMap<>(); - row.put("id", spec.id()); - row.put("type", spec.type()); - row.put("enabled", spec.enabled()); - Optional connector = resolve(spec.type()); - row.put("registered", connector.isPresent()); - row.put("scaffoldOnly", connector.map(c -> c.capabilities().scaffoldOnly()).orElse(true)); - if (connector.isEmpty()) { - row.put("error", "unknown_source_type"); + Map sourceDescription = new java.util.LinkedHashMap<>(); + // Compatibility boundary: the existing HTTP response still publishes legacy id/type keys. + sourceDescription.put("id", sourceSpec.sourceId()); + sourceDescription.put("type", sourceSpec.sourceType()); + sourceDescription.put("enabled", sourceSpec.enabled()); + Optional sourceConnector = resolve(sourceSpec.sourceType()); + sourceDescription.put("registered", sourceConnector.isPresent()); + sourceDescription.put( + "scaffoldOnly", + sourceConnector.map(connector -> connector.capabilities().scaffoldOnly()).orElse(true) + ); + if (sourceConnector.isEmpty()) { + sourceDescription.put("error", "unknown_source_type"); } - out.add(row); + sourceDescriptions.add(sourceDescription); } - return out; + return sourceDescriptions; } /** * Immutable source config entry (YAML list item). */ - public record SourceSpec(String id, String type, boolean enabled) { + public record SourceSpec(String sourceId, String sourceType, boolean enabled) { public SourceSpec { - if (id == null || id.isBlank()) { + if (sourceId == null || sourceId.isBlank()) { throw new IllegalArgumentException("source id required"); } - if (type == null || type.isBlank()) { + if (sourceType == null || sourceType.isBlank()) { throw new IllegalArgumentException("source type required"); } } From ba392728b5f344176903f65902181c59c6dd85c2 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 1 Sep 2026 21:49:57 +0900 Subject: [PATCH 03/36] refactor(cdc): name source configuration fields explicitly --- .../xtrmetl/cdc/config/XtrmetlProperties.java | 77 +++++++++++++++---- 1 file changed, 64 insertions(+), 13 deletions(-) diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/config/XtrmetlProperties.java b/cdc-service/src/main/java/com/xtrmetl/cdc/config/XtrmetlProperties.java index 067f5a70..b6875949 100644 --- a/cdc-service/src/main/java/com/xtrmetl/cdc/config/XtrmetlProperties.java +++ b/cdc-service/src/main/java/com/xtrmetl/cdc/config/XtrmetlProperties.java @@ -57,33 +57,84 @@ public void setSources(java.util.List sources) { } private static Source defaultPostgresSource() { - Source source = new Source(); - source.setId("pg-main"); - source.setType("postgres-debezium"); - source.setEnabled(true); - return source; + Source sourceConfiguration = new Source(); + sourceConfiguration.setSourceId("pg-main"); + sourceConfiguration.setSourceType("postgres-debezium"); + sourceConfiguration.setEnabled(true); + return sourceConfiguration; } } + /** + * Spring configuration adapter for one CDC source declaration. + * + *

The authoritative Java names are {@code sourceId}/{@code sourceType}. Legacy + * {@code id}/{@code type} bean accessors remain only so existing YAML continues to bind + * without a breaking configuration migration.

+ */ public static class Source { - private String id = "pg-main"; - private String type = "postgres-debezium"; + private String sourceId = "pg-main"; + private String sourceType = "postgres-debezium"; private boolean enabled = true; + public String getSourceId() { + return sourceId; + } + + public void setSourceId(String sourceId) { + this.sourceId = sourceId; + } + + public String getSourceType() { + return sourceType; + } + + public void setSourceType(String sourceType) { + this.sourceType = sourceType; + } + + /** + * Legacy Spring/YAML compatibility accessor for the historical {@code id} key. + * + * @return the configured CDC source identifier + * @deprecated internal callers must use {@link #getSourceId()} + */ + @Deprecated(forRemoval = false) public String getId() { - return id; + return sourceId; } - public void setId(String id) { - this.id = id; + /** + * Legacy Spring/YAML compatibility mutator for the historical {@code id} key. + * + * @param legacySourceId configured CDC source identifier + * @deprecated internal callers must use {@link #setSourceId(String)} + */ + @Deprecated(forRemoval = false) + public void setId(String legacySourceId) { + this.sourceId = legacySourceId; } + /** + * Legacy Spring/YAML compatibility accessor for the historical {@code type} key. + * + * @return the configured CDC source connector type + * @deprecated internal callers must use {@link #getSourceType()} + */ + @Deprecated(forRemoval = false) public String getType() { - return type; + return sourceType; } - public void setType(String type) { - this.type = type; + /** + * Legacy Spring/YAML compatibility mutator for the historical {@code type} key. + * + * @param legacySourceType configured CDC source connector type + * @deprecated internal callers must use {@link #setSourceType(String)} + */ + @Deprecated(forRemoval = false) + public void setType(String legacySourceType) { + this.sourceType = legacySourceType; } public boolean isEnabled() { From 9dad27bee1f41c9956e50f33fed5bb9aa5683911 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 1 Sep 2026 21:50:28 +0900 Subject: [PATCH 04/36] refactor(cdc): propagate semantic source names --- .../xtrmetl/cdc/controller/CdcController.java | 96 ++++++++++--------- 1 file changed, 52 insertions(+), 44 deletions(-) diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/controller/CdcController.java b/cdc-service/src/main/java/com/xtrmetl/cdc/controller/CdcController.java index 621af2df..f4a6c9dd 100644 --- a/cdc-service/src/main/java/com/xtrmetl/cdc/controller/CdcController.java +++ b/cdc-service/src/main/java/com/xtrmetl/cdc/controller/CdcController.java @@ -24,7 +24,7 @@ public class CdcController { private final CdcService cdcService; - private final XtrmetlProperties properties; + private final XtrmetlProperties xtrmetlProperties; private final CdcSourceRegistry sourceRegistry; private final CdcTargetRegistry targetRegistry; private final CdcSourceFactory sourceFactory; @@ -32,14 +32,14 @@ public class CdcController { public CdcController( CdcService cdcService, - XtrmetlProperties properties, + XtrmetlProperties xtrmetlProperties, CdcSourceRegistry sourceRegistry, CdcTargetRegistry targetRegistry, CdcSourceFactory sourceFactory, ReplicationSlotProbe replicationSlotProbe ) { this.cdcService = cdcService; - this.properties = properties; + this.xtrmetlProperties = xtrmetlProperties; this.sourceRegistry = sourceRegistry; this.targetRegistry = targetRegistry; this.sourceFactory = sourceFactory; @@ -48,64 +48,70 @@ public CdcController( @GetMapping("/status") @Observed(name = "cdc.status", contextualName = "cdc-status") - public ResponseEntity> status() { - Map body = new LinkedHashMap<>(cdcService.getStatus()); - body.put("replicaEnabled", properties.getReplica().isEnabled()); - body.put("replicaDdlEnabled", properties.getReplica().isDdlEnabled()); - body.put("replicaTopicPattern", properties.getReplica().getTopicPattern()); - body.put("replicaTables", properties.getReplica().getTables()); - body.put("replicationSlot", replicationSlotProbe.probeConfiguredSlot()); - body.put("configuredSources", sourceFactory.describeConfigured( - properties.getCdc().getSources().stream() - .map(s -> new CdcSourceFactory.SourceSpec(s.getId(), s.getType(), s.isEnabled())) + public ResponseEntity> cdcStatus() { + Map statusBody = new LinkedHashMap<>(cdcService.getStatus()); + statusBody.put("replicaEnabled", xtrmetlProperties.getReplica().isEnabled()); + statusBody.put("replicaDdlEnabled", xtrmetlProperties.getReplica().isDdlEnabled()); + statusBody.put("replicaTopicPattern", xtrmetlProperties.getReplica().getTopicPattern()); + statusBody.put("replicaTables", xtrmetlProperties.getReplica().getTables()); + statusBody.put("replicationSlot", replicationSlotProbe.probeConfiguredSlot()); + statusBody.put("configuredSources", sourceFactory.describeConfigured( + xtrmetlProperties.getCdc().getSources().stream() + .map(sourceConfiguration -> new CdcSourceFactory.SourceSpec( + sourceConfiguration.getSourceId(), + sourceConfiguration.getSourceType(), + sourceConfiguration.isEnabled() + )) .collect(Collectors.toList()) )); - body.put("registeredSources", sourceRegistry.all().stream() - .map(this::sourceEntry) + statusBody.put("registeredSources", sourceRegistry.all().stream() + .map(this::sourceRegistryEntry) .collect(Collectors.toList())); - body.put("registeredTargets", targetRegistry.all().stream() - .map(target -> { - Map entry = new LinkedHashMap<>(); - entry.put("id", target.id()); - entry.put("displayName", target.displayName()); - entry.put("scaffoldOnly", target.scaffoldOnly()); - return entry; + statusBody.put("registeredTargets", targetRegistry.all().stream() + .map(targetConnector -> { + Map targetEntry = new LinkedHashMap<>(); + targetEntry.put("id", targetConnector.id()); + targetEntry.put("displayName", targetConnector.displayName()); + targetEntry.put("scaffoldOnly", targetConnector.scaffoldOnly()); + return targetEntry; }) .collect(Collectors.toList())); - return ResponseEntity.ok(body); + return ResponseEntity.ok(statusBody); } @GetMapping("/sources") @Observed(name = "cdc.sources", contextualName = "cdc-sources") - public ResponseEntity>> sources() { + public ResponseEntity>> cdcSources() { return ResponseEntity.ok(sourceRegistry.all().stream() - .map(this::sourceEntry) + .map(this::sourceRegistryEntry) .collect(Collectors.toList())); } @GetMapping("/targets") @Observed(name = "cdc.targets", contextualName = "cdc-targets") - public ResponseEntity>> targets() { - List> body = targetRegistry.all().stream() - .map(target -> { - Map entry = new LinkedHashMap<>(); - entry.put("id", target.id()); - entry.put("displayName", target.displayName()); - entry.put("scaffoldOnly", target.scaffoldOnly()); - return entry; + public ResponseEntity>> cdcTargets() { + List> targetEntries = targetRegistry.all().stream() + .map(targetConnector -> { + Map targetEntry = new LinkedHashMap<>(); + targetEntry.put("id", targetConnector.id()); + targetEntry.put("displayName", targetConnector.displayName()); + targetEntry.put("scaffoldOnly", targetConnector.scaffoldOnly()); + return targetEntry; }) .collect(Collectors.toList()); - return ResponseEntity.ok(body); + return ResponseEntity.ok(targetEntries); } - private Map sourceEntry(com.xtrmetl.cdc.spi.CdcSourceConnector source) { - Map entry = new LinkedHashMap<>(); - entry.put("id", source.id()); - entry.put("displayName", source.displayName()); - entry.put("engine", source.capabilities().engine()); - entry.put("databases", source.capabilities().databases()); - entry.put("scaffoldOnly", source.capabilities().scaffoldOnly()); - return entry; + private Map sourceRegistryEntry( + com.xtrmetl.cdc.spi.CdcSourceConnector sourceConnector + ) { + Map sourceEntry = new LinkedHashMap<>(); + sourceEntry.put("id", sourceConnector.id()); + sourceEntry.put("displayName", sourceConnector.displayName()); + sourceEntry.put("engine", sourceConnector.capabilities().engine()); + sourceEntry.put("databases", sourceConnector.capabilities().databases()); + sourceEntry.put("scaffoldOnly", sourceConnector.capabilities().scaffoldOnly()); + return sourceEntry; } @PostMapping("/start") @@ -121,8 +127,10 @@ public ResponseEntity stopCdc() { try { cdcService.stop(); return ResponseEntity.ok("CDC process stopped"); - } catch (IOException e) { - return ResponseEntity.internalServerError().body("Error stopping CDC process: " + e.getMessage()); + } catch (IOException stopFailure) { + return ResponseEntity.internalServerError().body( + "Error stopping CDC process: " + stopFailure.getMessage() + ); } } } From c734afe822a35befabc609a45359d7fd3836381e Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 1 Sep 2026 21:51:01 +0900 Subject: [PATCH 05/36] test(cdc): align controller semantic names --- .../cdc/controller/CdcControllerTest.java | 60 +++++++++++-------- 1 file changed, 34 insertions(+), 26 deletions(-) diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/controller/CdcControllerTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/controller/CdcControllerTest.java index 08cf2f5d..bab8517a 100644 --- a/cdc-service/src/test/java/com/xtrmetl/cdc/controller/CdcControllerTest.java +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/controller/CdcControllerTest.java @@ -28,7 +28,7 @@ class CdcControllerTest { private CdcService cdcService; - private XtrmetlProperties properties; + private XtrmetlProperties xtrmetlProperties; private CdcSourceRegistry sourceRegistry; private CdcTargetRegistry targetRegistry; private CdcSourceFactory sourceFactory; @@ -38,7 +38,7 @@ class CdcControllerTest { @BeforeEach void setUp() { cdcService = mock(CdcService.class); - properties = new XtrmetlProperties(); + xtrmetlProperties = new XtrmetlProperties(); sourceRegistry = new CdcSourceRegistry(); targetRegistry = new CdcTargetRegistry(); sourceFactory = new CdcSourceFactory(sourceRegistry); @@ -52,7 +52,7 @@ void setUp() { )); cdcController = new CdcController( cdcService, - properties, + xtrmetlProperties, sourceRegistry, targetRegistry, sourceFactory, @@ -96,44 +96,52 @@ void testStatusIncludesSlotSourcesAndTargets() { serviceStatus.put("product", "mightyETL"); when(cdcService.getStatus()).thenReturn(serviceStatus); - ResponseEntity> response = cdcController.status(); + ResponseEntity> response = cdcController.cdcStatus(); assertEquals(HttpStatus.OK, response.getStatusCode()); - Map body = response.getBody(); - assertEquals(false, body.get("running")); - assertEquals("mightyETL", body.get("product")); - assertEquals(false, body.get("replicaEnabled")); - assertTrue(body.containsKey("registeredSources")); - assertTrue(body.containsKey("registeredTargets")); - assertTrue(body.containsKey("configuredSources")); - assertTrue(body.containsKey("replicationSlot")); + Map statusBody = response.getBody(); + assertEquals(false, statusBody.get("running")); + assertEquals("mightyETL", statusBody.get("product")); + assertEquals(false, statusBody.get("replicaEnabled")); + assertTrue(statusBody.containsKey("registeredSources")); + assertTrue(statusBody.containsKey("registeredTargets")); + assertTrue(statusBody.containsKey("configuredSources")); + assertTrue(statusBody.containsKey("replicationSlot")); @SuppressWarnings("unchecked") - Map slot = (Map) body.get("replicationSlot"); - assertEquals(true, slot.get("found")); + Map replicationSlot = + (Map) statusBody.get("replicationSlot"); + assertEquals(true, replicationSlot.get("found")); @SuppressWarnings("unchecked") - List> sources = (List>) body.get("registeredSources"); - assertFalse(sources.isEmpty()); - assertEquals("postgres-debezium", sources.get(0).get("id")); + List> sourceEntries = + (List>) statusBody.get("registeredSources"); + assertFalse(sourceEntries.isEmpty()); + assertEquals("postgres-debezium", sourceEntries.get(0).get("id")); } @Test void testSourcesListsPostgresDebezium() { - ResponseEntity>> response = cdcController.sources(); + ResponseEntity>> response = cdcController.cdcSources(); assertEquals(HttpStatus.OK, response.getStatusCode()); - List> body = response.getBody(); - assertFalse(body.isEmpty()); - assertTrue(body.stream().anyMatch(s -> "postgres-debezium".equals(s.get("id")))); + List> sourceEntries = response.getBody(); + assertFalse(sourceEntries.isEmpty()); + assertTrue(sourceEntries.stream().anyMatch( + sourceEntry -> "postgres-debezium".equals(sourceEntry.get("id")) + )); } @Test void testTargetsListsKafkaAndJdbcReplica() { - ResponseEntity>> response = cdcController.targets(); + ResponseEntity>> response = cdcController.cdcTargets(); assertEquals(HttpStatus.OK, response.getStatusCode()); - List> body = response.getBody(); - assertEquals(2, body.size()); - assertTrue(body.stream().anyMatch(t -> "kafka".equals(t.get("id")))); - assertTrue(body.stream().anyMatch(t -> "jdbc-replica".equals(t.get("id")))); + List> targetEntries = response.getBody(); + assertEquals(2, targetEntries.size()); + assertTrue(targetEntries.stream().anyMatch( + targetEntry -> "kafka".equals(targetEntry.get("id")) + )); + assertTrue(targetEntries.stream().anyMatch( + targetEntry -> "jdbc-replica".equals(targetEntry.get("id")) + )); } } From c0fa589a6e66cf03b90e26df80dc68fc7bbb8744 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 1 Sep 2026 21:53:35 +0900 Subject: [PATCH 06/36] test(cdc): preserve legacy source config binding --- .../XtrmetlPropertiesSecurityDefaultTest.java | 46 +++++++++++++++++-- 1 file changed, 42 insertions(+), 4 deletions(-) diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/config/XtrmetlPropertiesSecurityDefaultTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/config/XtrmetlPropertiesSecurityDefaultTest.java index 73849a9f..39473019 100644 --- a/cdc-service/src/test/java/com/xtrmetl/cdc/config/XtrmetlPropertiesSecurityDefaultTest.java +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/config/XtrmetlPropertiesSecurityDefaultTest.java @@ -1,24 +1,62 @@ package com.xtrmetl.cdc.config; import org.junit.jupiter.api.Test; +import org.springframework.boot.context.properties.bind.Bindable; +import org.springframework.boot.context.properties.bind.Binder; +import org.springframework.boot.context.properties.source.MapConfigurationPropertySource; + +import java.util.Map; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; /** - * Guards the secure standalone default exposed by the CDC replica configuration object. + * Guards secure CDC defaults and the legacy-to-semantic source configuration boundary. */ class XtrmetlPropertiesSecurityDefaultTest { @Test void replicaDdlRemainsDisabledAndUsesWhitelistValidationByDefault() { - XtrmetlProperties properties = new XtrmetlProperties(); + XtrmetlProperties xtrmetlProperties = new XtrmetlProperties(); - assertFalse(properties.getReplica().isDdlEnabled(), "DDL replication must remain disabled by default"); + assertFalse( + xtrmetlProperties.getReplica().isDdlEnabled(), + "DDL replication must remain disabled by default" + ); assertEquals( "whitelist", - properties.getReplica().getDdlValidationMode(), + xtrmetlProperties.getReplica().getDdlValidationMode(), "the Java configuration object must match the deployable and metadata secure default" ); } + + @Test + void defaultCdcSourceUsesSemanticJavaIdentifiers() { + XtrmetlProperties xtrmetlProperties = new XtrmetlProperties(); + XtrmetlProperties.Source sourceConfiguration = + xtrmetlProperties.getCdc().getSources().getFirst(); + + assertEquals("pg-main", sourceConfiguration.getSourceId()); + assertEquals("postgres-debezium", sourceConfiguration.getSourceType()); + } + + @Test + void legacySourceKeysBindIntoSemanticJavaIdentifiers() { + MapConfigurationPropertySource configurationSource = + new MapConfigurationPropertySource(Map.of( + "xtrmetl.cdc.sources[0].id", "pg-legacy", + "xtrmetl.cdc.sources[0].type", "mysql-debezium", + "xtrmetl.cdc.sources[0].enabled", "false" + )); + + XtrmetlProperties xtrmetlProperties = new Binder(configurationSource) + .bind("xtrmetl", Bindable.of(XtrmetlProperties.class)) + .orElseThrow(); + XtrmetlProperties.Source sourceConfiguration = + xtrmetlProperties.getCdc().getSources().getFirst(); + + assertEquals("pg-legacy", sourceConfiguration.getSourceId()); + assertEquals("mysql-debezium", sourceConfiguration.getSourceType()); + assertFalse(sourceConfiguration.isEnabled()); + } } From 3e1bc21562c9677dba9b1e0cd550954fe6c9ef5c Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 1 Sep 2026 21:55:46 +0900 Subject: [PATCH 07/36] docs(cdc): record semantic source naming boundary --- .../cdc-source-semantic-identifiers.md | 42 +++++++++++++++++++ 1 file changed, 42 insertions(+) create mode 100644 docs/doctoring/cdc-source-semantic-identifiers.md diff --git a/docs/doctoring/cdc-source-semantic-identifiers.md b/docs/doctoring/cdc-source-semantic-identifiers.md new file mode 100644 index 00000000..fde67b23 --- /dev/null +++ b/docs/doctoring/cdc-source-semantic-identifiers.md @@ -0,0 +1,42 @@ +# CDC source semantic identifier doctoring + +## Decision + +The CDC source configuration bounded context owns source identity and connector-type vocabulary. Internal Java identifiers therefore use the explicit terms `sourceId` and `sourceType` rather than the generic one-word `id` and `type` names. The Java casing remains idiomatic camelCase; the decision is about semantic specificity, not forcing snake_case into Java. + +`CdcSourceFactory.SourceSpec` is the authoritative internal value object for one declared source. Its record components are `sourceId`, `sourceType`, and `enabled`. Factory parameters and local variables use source-specific terms (`sourceRegistry`, `sourceSpecs`, `sourceDescription`, `sourceConnector`) so ownership remains visible at the call site. + +## Compatibility boundary + +Existing operators already configure `xtrmetl.cdc.sources[*].id` and `xtrmetl.cdc.sources[*].type`, and the CDC status resource already emits configured-source objects containing `id` and `type`. This change therefore treats those historical keys as an anti-corruption/compatibility boundary rather than silently changing the wire/config contract. + +`XtrmetlProperties.Source` stores values internally as `sourceId` and `sourceType`. The historical `getId`/`setId` and `getType`/`setType` JavaBean accessors remain only so Spring can continue binding the established configuration keys. Repository-owned production callers use `getSourceId`/`getSourceType`. Focused binder coverage proves that legacy `id`/`type` configuration still populates the semantic internal fields. + +`CdcSourceFactory.describeConfigured` likewise keeps the established output keys `id` and `type`, while all internal record access is through `sourceId()` and `sourceType()`. The compatibility regression explicitly fails if the implementation starts leaking new Java member names into the existing HTTP payload. + +## DDD traceability + +- **Bounded context:** Change Data Capture source configuration and discovery. +- **Ubiquitous language:** CDC source, source identifier, source connector type, source registry, configured source, registered source. +- **Value object:** `CdcSourceFactory.SourceSpec` represents one immutable source declaration. +- **Domain service:** `CdcSourceFactory` validates source identity uniqueness and resolves source connector types. +- **Invariant:** two configured source declarations cannot share the same `sourceId`. +- **Invariant:** `sourceId` and `sourceType` must be nonblank before a declaration enters the configured-source description pipeline. +- **Compatibility invariant:** established operator configuration and HTTP keys remain `id`/`type` until a separately versioned breaking contract intentionally replaces them. +- **Persistence:** no database row, column, index, migration, Kafka record, or Debezium offset changes in this repair. + +## Verification contract + +Focused tests require semantic record accessors, duplicate-source rejection, repeated connector types with distinct source identities, unchanged legacy HTTP keys, semantic Java defaults, and Spring binding of the historical `id`/`type` keys into the new internal fields. The full Maven/CI, dependency review, SBOM, SAST, and Security Scan gates remain authoritative on the unchanged resulting head. + +The regression-only head was intentionally superseded quickly by ordinary non-force implementation commits, so its hosted jobs were cancelled rather than claimed as RED evidence. The regression itself is retained in history and remains executable on the final branch; no cancelled or predecessor run is counted as merge evidence. + +## Research basis + +The naming rule is intentionally semantic rather than mechanical. Schankin et al. found that descriptive compound identifiers helped experienced Java developers locate semantic defects faster than shorter, less descriptive names. Feitelson et al. model naming as selecting the concepts a name should communicate, choosing words for those concepts, and constructing the identifier; explicitly applying that model produced names judged superior to unconstrained choices. These findings support encoding the owned concepts (`source` + `id`, `source` + `type`) while retaining the host language's normal casing and allowing concise names where context itself already supplies the meaning. + +### References + +Feitelson, D. G., Mizrahi, A., Noy, N., Ben Shabat, A., Eliyahu, O., & Sheffer, R. (2022). How developers choose names. *IEEE Transactions on Software Engineering, 48*(1), 37–52. https://doi.org/10.1109/TSE.2020.2976920 + +Schankin, A., Berger, A., Holt, D. V., Hofmeister, J. C., Riedel, T., & Beigl, M. (2018). Descriptive compound identifier names improve source code comprehension. In *Proceedings of the 26th Conference on Program Comprehension* (pp. 31–40). Association for Computing Machinery. https://doi.org/10.1145/3196321.3196332 From 949b04aafdaa895ff273cfb91d2b7ff4bc5dacfc Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 02:49:14 +0900 Subject: [PATCH 08/36] test(cdc): require semantic connector identities --- .../spi/CdcConnectorSemanticIdentityTest.java | 27 +++++++++++++++++++ 1 file changed, 27 insertions(+) create mode 100644 cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcConnectorSemanticIdentityTest.java diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcConnectorSemanticIdentityTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcConnectorSemanticIdentityTest.java new file mode 100644 index 00000000..1e1c69b3 --- /dev/null +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcConnectorSemanticIdentityTest.java @@ -0,0 +1,27 @@ +package com.xtrmetl.cdc.spi; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * Guards bounded-context-specific identity accessors on the organization-owned CDC SPI. + */ +class CdcConnectorSemanticIdentityTest { + + @Test + void sourceConnectorExposesSourceIdentityBySemanticName() { + CdcSourceConnector sourceConnector = new PostgresDebeziumCdcSource(); + + assertEquals("postgres-debezium", sourceConnector.sourceId()); + } + + @Test + void targetConnectorsExposeTargetIdentityBySemanticName() { + CdcTargetConnector kafkaTargetConnector = new KafkaCdcTargetConnector(); + CdcTargetConnector jdbcTargetConnector = new JdbcReplicaCdcTargetConnector(); + + assertEquals("kafka", kafkaTargetConnector.targetId()); + assertEquals("jdbc-replica", jdbcTargetConnector.targetId()); + } +} From e941a174dec5a68b2d25d40553a10f83ffe432ee Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 02:53:29 +0900 Subject: [PATCH 09/36] refactor(cdc): expose semantic source identity --- .../xtrmetl/cdc/spi/CdcSourceConnector.java | 24 +++++++++++++++++-- 1 file changed, 22 insertions(+), 2 deletions(-) diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcSourceConnector.java b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcSourceConnector.java index 515047a6..98e661fd 100644 --- a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcSourceConnector.java +++ b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcSourceConnector.java @@ -8,6 +8,26 @@ */ public interface CdcSourceConnector extends AutoCloseable { + /** + * Returns the bounded-context-specific CDC source identifier. + * + *

New organization-owned callers must use this semantic accessor. The generic + * {@link #id()} method remains only as an SPI compatibility boundary for existing + * external connector implementations and callers.

+ * + * @return exact CDC source identifier + */ + default String sourceId() { + return id(); + } + + /** + * Legacy compatibility accessor for the historical generic connector identifier. + * + * @return exact CDC source identifier + * @deprecated organization-owned callers must use {@link #sourceId()} + */ + @Deprecated(forRemoval = false) String id(); String displayName(); @@ -17,12 +37,12 @@ public interface CdcSourceConnector extends AutoCloseable { /** * Validate source configuration (host, slot, credentials, include lists). */ - void validate(Map config); + void validate(Map sourceConfig); /** * Begin capturing changes. Implementations publish through the service pipeline. */ - void start(Map config); + void start(Map sourceConfig); /** * Stop capture gracefully. From 15a3ccd6838340d0447f0b43ea7f9207230f15f8 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 02:53:47 +0900 Subject: [PATCH 10/36] refactor(cdc): expose semantic target identity --- .../xtrmetl/cdc/spi/CdcTargetConnector.java | 24 +++++++++++++++++-- 1 file changed, 22 insertions(+), 2 deletions(-) diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcTargetConnector.java b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcTargetConnector.java index 07466023..0469b347 100644 --- a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcTargetConnector.java +++ b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcTargetConnector.java @@ -34,6 +34,26 @@ record Capabilities( ) { } + /** + * Returns the bounded-context-specific CDC target identifier. + * + *

New organization-owned callers must use this semantic accessor. The generic + * {@link #id()} method remains only as an SPI compatibility boundary for existing + * external connector implementations and callers.

+ * + * @return exact CDC target identifier + */ + default String targetId() { + return id(); + } + + /** + * Legacy compatibility accessor for the historical generic connector identifier. + * + * @return exact CDC target identifier + * @deprecated organization-owned callers must use {@link #targetId()} + */ + @Deprecated(forRemoval = false) String id(); String displayName(); @@ -50,12 +70,12 @@ record Capabilities( */ Capabilities capabilities(); - void validate(Map config); + void validate(Map targetConfig); /** * Apply a batch of canonical change records. */ - void write(List batch); + void write(List changeBatch); @Override default void close() { From d9fdf2ad3e7c59c47d1c6a835c1677c7d782f7af Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 02:54:06 +0900 Subject: [PATCH 11/36] refactor(cdc): name scaffold source identity explicitly --- .../cdc/spi/AbstractScaffoldCdcSource.java | 38 +++++++++++++------ 1 file changed, 26 insertions(+), 12 deletions(-) diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/AbstractScaffoldCdcSource.java b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/AbstractScaffoldCdcSource.java index e493fbad..2e50829a 100644 --- a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/AbstractScaffoldCdcSource.java +++ b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/AbstractScaffoldCdcSource.java @@ -9,21 +9,35 @@ */ public abstract class AbstractScaffoldCdcSource implements CdcSourceConnector { - private final String id; + private final String sourceId; private final String displayName; - private final String engine; - private final Set databases; + private final String sourceEngine; + private final Set supportedDatabases; - protected AbstractScaffoldCdcSource(String id, String displayName, String engine, Set databases) { - this.id = Objects.requireNonNull(id, "id"); + protected AbstractScaffoldCdcSource( + String sourceId, + String displayName, + String sourceEngine, + Set supportedDatabases + ) { + this.sourceId = Objects.requireNonNull(sourceId, "sourceId"); this.displayName = Objects.requireNonNull(displayName, "displayName"); - this.engine = Objects.requireNonNull(engine, "engine"); - this.databases = Set.copyOf(databases); + this.sourceEngine = Objects.requireNonNull(sourceEngine, "sourceEngine"); + this.supportedDatabases = Set.copyOf(supportedDatabases); } @Override + public final String sourceId() { + return sourceId; + } + + /** + * @deprecated compatibility alias; organization-owned callers use {@link #sourceId()} + */ + @Override + @Deprecated(forRemoval = false) public final String id() { - return id; + return sourceId(); } @Override @@ -33,16 +47,16 @@ public final String displayName() { @Override public final SourceCapabilities capabilities() { - return new SourceCapabilities(engine, databases, true); + return new SourceCapabilities(sourceEngine, supportedDatabases, true); } @Override - public void validate(Map config) { - Objects.requireNonNull(config, "config"); + public void validate(Map sourceConfig) { + Objects.requireNonNull(sourceConfig, "sourceConfig"); } @Override - public final void start(Map config) { + public final void start(Map sourceConfig) { throw new UnsupportedOperationException( displayName + " is a scaffold source only (no connector dependency wired). " + "See docs/cdc/any-to-any-cdc.md" From 52db1faa594aedc5659d447ed60427b83ddd896d Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 02:54:30 +0900 Subject: [PATCH 12/36] refactor(cdc): make PostgreSQL source identity explicit --- .../cdc/spi/PostgresDebeziumCdcSource.java | 65 ++++++++++++------- 1 file changed, 40 insertions(+), 25 deletions(-) diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/PostgresDebeziumCdcSource.java b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/PostgresDebeziumCdcSource.java index 673a4578..67173d9c 100644 --- a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/PostgresDebeziumCdcSource.java +++ b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/PostgresDebeziumCdcSource.java @@ -21,17 +21,23 @@ @Component public final class PostgresDebeziumCdcSource implements CdcSourceConnector { - public static final String ID = "postgres-debezium"; + public static final String SOURCE_ID = "postgres-debezium"; - private final ObjectProvider cdcService; + /** + * @deprecated compatibility alias; organization-owned callers use {@link #SOURCE_ID} + */ + @Deprecated(forRemoval = false) + public static final String ID = SOURCE_ID; + + private final ObjectProvider cdcServiceProvider; /** * Creates the PostgreSQL CDC adapter backed by the deployment-configured live service. * - * @param cdcService provider for the live CDC service + * @param cdcServiceProvider provider for the live CDC service */ - public PostgresDebeziumCdcSource(ObjectProvider cdcService) { - this.cdcService = cdcService; + public PostgresDebeziumCdcSource(ObjectProvider cdcServiceProvider) { + this.cdcServiceProvider = cdcServiceProvider; } /** @@ -62,8 +68,17 @@ public CdcService getIfUnique() { } @Override + public String sourceId() { + return SOURCE_ID; + } + + /** + * @deprecated compatibility alias; organization-owned callers use {@link #sourceId()} + */ + @Override + @Deprecated(forRemoval = false) public String id() { - return ID; + return sourceId(); } @Override @@ -82,17 +97,17 @@ public SourceCapabilities capabilities() { *

The live PostgreSQL capture service is configured by the deployment, not by this SPI call. * An empty map is therefore the only valid value.

* - * @param config per-call settings; must be non-null and empty - * @throws IllegalArgumentException when {@code config} is null or contains any entry + * @param sourceConfig per-call settings; must be non-null and empty + * @throws IllegalArgumentException when {@code sourceConfig} is null or contains any entry */ @Override - public void validate(Map config) { - if (config == null) { - throw new IllegalArgumentException("config must not be null"); + public void validate(Map sourceConfig) { + if (sourceConfig == null) { + throw new IllegalArgumentException("sourceConfig must not be null"); } - if (!config.isEmpty()) { + if (!sourceConfig.isEmpty()) { throw new IllegalArgumentException( - "postgres-debezium uses deployment-owned configuration; per-call config must be empty" + "postgres-debezium uses deployment-owned configuration; per-call sourceConfig must be empty" ); } } @@ -100,32 +115,32 @@ public void validate(Map config) { /** * Starts the deployment-configured PostgreSQL CDC service. * - * @param config per-call settings; must be non-null and empty - * @throws IllegalArgumentException when {@code config} is null or contains any entry + * @param sourceConfig per-call settings; must be non-null and empty + * @throws IllegalArgumentException when {@code sourceConfig} is null or contains any entry * @throws IllegalStateException when the live {@link CdcService} is unavailable */ @Override - public void start(Map config) { - validate(config); - CdcService service = cdcService.getIfAvailable(); - if (service == null) { + public void start(Map sourceConfig) { + validate(sourceConfig); + CdcService liveCdcService = cdcServiceProvider.getIfAvailable(); + if (liveCdcService == null) { throw new IllegalStateException( "CdcService is not available; cannot start postgres-debezium via SPI" ); } - service.start(); + liveCdcService.start(); } @Override public void stop() { - CdcService service = cdcService.getIfAvailable(); - if (service == null) { + CdcService liveCdcService = cdcServiceProvider.getIfAvailable(); + if (liveCdcService == null) { return; } try { - service.stop(); - } catch (IOException e) { - throw new IllegalStateException("Error stopping CDC via SPI", e); + liveCdcService.stop(); + } catch (IOException stopFailure) { + throw new IllegalStateException("Error stopping CDC via SPI", stopFailure); } } } From 0a3b0f9d55b3b209ed7c24d7ac8f73c01bccf949 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 02:54:51 +0900 Subject: [PATCH 13/36] refactor(cdc): make Kafka target identity explicit --- .../cdc/spi/KafkaCdcTargetConnector.java | 27 ++++++++++++++----- 1 file changed, 21 insertions(+), 6 deletions(-) diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/KafkaCdcTargetConnector.java b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/KafkaCdcTargetConnector.java index 5214e310..2e38a33a 100644 --- a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/KafkaCdcTargetConnector.java +++ b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/KafkaCdcTargetConnector.java @@ -9,11 +9,26 @@ */ public final class KafkaCdcTargetConnector implements CdcTargetConnector { - public static final String ID = "kafka"; + public static final String TARGET_ID = "kafka"; + /** + * @deprecated compatibility alias; organization-owned callers use {@link #TARGET_ID} + */ + @Deprecated(forRemoval = false) + public static final String ID = TARGET_ID; + + @Override + public String targetId() { + return TARGET_ID; + } + + /** + * @deprecated compatibility alias; organization-owned callers use {@link #targetId()} + */ @Override + @Deprecated(forRemoval = false) public String id() { - return ID; + return targetId(); } @Override @@ -37,14 +52,14 @@ public Capabilities capabilities() { } @Override - public void validate(Map config) { - if (config == null) { - throw new IllegalArgumentException("config must not be null"); + public void validate(Map targetConfig) { + if (targetConfig == null) { + throw new IllegalArgumentException("targetConfig must not be null"); } } @Override - public void write(List batch) { + public void write(List changeBatch) { throw new UnsupportedOperationException( "Live Kafka publish still uses raw Debezium envelopes via CdcService; " + "canonical-record routing is not wired yet. See docs/cdc/any-to-any-cdc.md" From 911b88725f0982a84929e673e12448de037643a4 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 02:55:04 +0900 Subject: [PATCH 14/36] refactor(cdc): make JDBC target identity explicit --- .../spi/JdbcReplicaCdcTargetConnector.java | 27 ++++++++++++++----- 1 file changed, 21 insertions(+), 6 deletions(-) diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/JdbcReplicaCdcTargetConnector.java b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/JdbcReplicaCdcTargetConnector.java index 26654369..96055e2c 100644 --- a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/JdbcReplicaCdcTargetConnector.java +++ b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/JdbcReplicaCdcTargetConnector.java @@ -8,11 +8,26 @@ */ public final class JdbcReplicaCdcTargetConnector implements CdcTargetConnector { - public static final String ID = "jdbc-replica"; + public static final String TARGET_ID = "jdbc-replica"; + /** + * @deprecated compatibility alias; organization-owned callers use {@link #TARGET_ID} + */ + @Deprecated(forRemoval = false) + public static final String ID = TARGET_ID; + + @Override + public String targetId() { + return TARGET_ID; + } + + /** + * @deprecated compatibility alias; organization-owned callers use {@link #targetId()} + */ @Override + @Deprecated(forRemoval = false) public String id() { - return ID; + return targetId(); } @Override @@ -36,14 +51,14 @@ public Capabilities capabilities() { } @Override - public void validate(Map config) { - if (config == null) { - throw new IllegalArgumentException("config must not be null"); + public void validate(Map targetConfig) { + if (targetConfig == null) { + throw new IllegalArgumentException("targetConfig must not be null"); } } @Override - public void write(List batch) { + public void write(List changeBatch) { throw new UnsupportedOperationException( "Replica apply is owned by ProcessedDataReplicaApplier (table processed_data only); " + "SPI write path not wired. See docs/cdc/ops-and-reliability.md" From ab43feb908b748fd3da1366f42d2d83f53084d17 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 02:55:33 +0900 Subject: [PATCH 15/36] refactor(cdc): use semantic source ids in registry --- .../xtrmetl/cdc/spi/CdcSourceRegistry.java | 52 +++++++++---------- 1 file changed, 26 insertions(+), 26 deletions(-) diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcSourceRegistry.java b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcSourceRegistry.java index 1a4e2ceb..fc5351bf 100644 --- a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcSourceRegistry.java +++ b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcSourceRegistry.java @@ -15,25 +15,25 @@ * Registry of CDC source connector types discovered as Spring beans, plus a safe * fallback for unit tests without a Spring context. * - *

Connector identifiers are configuration authority. Registration therefore fails closed for + *

Source connector identifiers are configuration authority. Registration therefore fails closed for * null connectors, blank identifiers, and duplicate identifiers instead of allowing bean order to * replace the selected implementation.

*/ @Component public class CdcSourceRegistry { - private final Map byId = new LinkedHashMap<>(); + private final Map sourceConnectorsById = new LinkedHashMap<>(); /** * Creates a registry from Spring-discovered source connectors. * - * @param connectors ordered provider of source connector beans - * @throws IllegalArgumentException when a discovered connector has an invalid or duplicate id + * @param sourceConnectors ordered provider of source connector beans + * @throws IllegalArgumentException when a discovered connector has an invalid or duplicate source id */ @Autowired - public CdcSourceRegistry(ObjectProvider connectors) { - connectors.orderedStream().forEach(this::register); - if (byId.isEmpty()) { + public CdcSourceRegistry(ObjectProvider sourceConnectors) { + sourceConnectors.orderedStream().forEach(this::register); + if (sourceConnectorsById.isEmpty()) { // Unit tests / non-Spring construction register(new PostgresDebeziumCdcSource()); } @@ -42,14 +42,14 @@ public CdcSourceRegistry(ObjectProvider connectors) { /** * Creates a registry from an explicit connector list, primarily for standalone use and tests. * - * @param connectors source connectors to register; a null list means no explicit connectors - * @throws IllegalArgumentException when a connector has an invalid or duplicate id + * @param sourceConnectors source connectors to register; a null list means no explicit connectors + * @throws IllegalArgumentException when a connector has an invalid or duplicate source id */ - public CdcSourceRegistry(List connectors) { - if (connectors != null) { - connectors.forEach(this::register); + public CdcSourceRegistry(List sourceConnectors) { + if (sourceConnectors != null) { + sourceConnectors.forEach(this::register); } - if (byId.isEmpty()) { + if (sourceConnectorsById.isEmpty()) { register(new PostgresDebeziumCdcSource()); } } @@ -64,32 +64,32 @@ public CdcSourceRegistry() { /** * Registers one source connector without allowing existing configuration identity to be replaced. * - * @param connector source connector to register - * @throws IllegalArgumentException when the connector is null, its id is blank, or its id is + * @param sourceConnector source connector to register + * @throws IllegalArgumentException when the connector is null, its source id is blank, or its source id is * already registered */ - public final void register(CdcSourceConnector connector) { - if (connector == null) { + public final void register(CdcSourceConnector sourceConnector) { + if (sourceConnector == null) { throw new IllegalArgumentException("CDC source connector must not be null"); } - String id = Objects.requireNonNullElse(connector.id(), ""); - if (id.isBlank()) { + String sourceId = Objects.requireNonNullElse(sourceConnector.sourceId(), ""); + if (sourceId.isBlank()) { throw new IllegalArgumentException("CDC source connector id must not be blank"); } - CdcSourceConnector previous = byId.putIfAbsent(id, connector); - if (previous != null) { - throw new IllegalArgumentException("Duplicate CDC source connector id: " + id); + CdcSourceConnector previousConnector = sourceConnectorsById.putIfAbsent(sourceId, sourceConnector); + if (previousConnector != null) { + throw new IllegalArgumentException("Duplicate CDC source connector id: " + sourceId); } } /** * Finds a source connector by its exact configuration identifier. * - * @param id exact connector identifier + * @param sourceId exact source connector identifier * @return the registered connector, or empty when the identifier is unknown */ - public Optional find(String id) { - return Optional.ofNullable(byId.get(id)); + public Optional find(String sourceId) { + return Optional.ofNullable(sourceConnectorsById.get(sourceId)); } /** @@ -98,6 +98,6 @@ public Optional find(String id) { * @return immutable connector collection detached from registry mutation authority */ public Collection all() { - return List.copyOf(byId.values()); + return List.copyOf(sourceConnectorsById.values()); } } From 12bba4fe1f07504129d3bb00722091106353d70d Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 02:55:51 +0900 Subject: [PATCH 16/36] refactor(cdc): use semantic target ids in registry --- .../xtrmetl/cdc/spi/CdcTargetRegistry.java | 32 +++++++++---------- 1 file changed, 16 insertions(+), 16 deletions(-) diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcTargetRegistry.java b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcTargetRegistry.java index 746c00db..5f96489f 100644 --- a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcTargetRegistry.java +++ b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcTargetRegistry.java @@ -12,13 +12,13 @@ /** * Registry of CDC target connector types such as Kafka and JDBC replica targets. * - *

Connector identifiers are configuration authority. Invalid registration is rejected so + *

Target connector identifiers are configuration authority. Invalid registration is rejected so * registration order cannot silently replace a selected connector implementation.

*/ @Component public class CdcTargetRegistry { - private final Map byId = new LinkedHashMap<>(); + private final Map targetConnectorsById = new LinkedHashMap<>(); /** Creates a registry containing the built-in Kafka and JDBC replica target connectors. */ public CdcTargetRegistry() { @@ -27,33 +27,33 @@ public CdcTargetRegistry() { } /** - * Registers one target connector without replacing an existing connector with the same id. + * Registers one target connector without replacing an existing connector with the same target id. * - * @param connector target connector to register - * @throws IllegalArgumentException when the connector is null, its id is blank, or its id is already registered + * @param targetConnector target connector to register + * @throws IllegalArgumentException when the connector is null, its target id is blank, or its target id is already registered */ - public final void register(CdcTargetConnector connector) { - if (connector == null) { + public final void register(CdcTargetConnector targetConnector) { + if (targetConnector == null) { throw new IllegalArgumentException("CDC target connector must not be null"); } - String id = Objects.requireNonNullElse(connector.id(), ""); - if (id.isBlank()) { + String targetId = Objects.requireNonNullElse(targetConnector.targetId(), ""); + if (targetId.isBlank()) { throw new IllegalArgumentException("CDC target connector id must not be blank"); } - CdcTargetConnector previous = byId.putIfAbsent(id, connector); - if (previous != null) { - throw new IllegalArgumentException("Duplicate CDC target connector id: " + id); + CdcTargetConnector previousConnector = targetConnectorsById.putIfAbsent(targetId, targetConnector); + if (previousConnector != null) { + throw new IllegalArgumentException("Duplicate CDC target connector id: " + targetId); } } /** * Finds a target connector by its exact configuration identifier. * - * @param id exact connector identifier + * @param targetId exact target connector identifier * @return the registered connector, or empty when the identifier is unknown */ - public Optional find(String id) { - return Optional.ofNullable(byId.get(id)); + public Optional find(String targetId) { + return Optional.ofNullable(targetConnectorsById.get(targetId)); } /** @@ -62,6 +62,6 @@ public Optional find(String id) { * @return immutable connector collection detached from registry mutation authority */ public Collection all() { - return List.copyOf(byId.values()); + return List.copyOf(targetConnectorsById.values()); } } From 875b8c7bb465e54cb23d0f8b7719c4f7aa9ddc6f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 02:56:12 +0900 Subject: [PATCH 17/36] refactor(cdc): use semantic connector ids internally --- .../main/java/com/xtrmetl/cdc/controller/CdcController.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/controller/CdcController.java b/cdc-service/src/main/java/com/xtrmetl/cdc/controller/CdcController.java index f4a6c9dd..f88dccdf 100644 --- a/cdc-service/src/main/java/com/xtrmetl/cdc/controller/CdcController.java +++ b/cdc-service/src/main/java/com/xtrmetl/cdc/controller/CdcController.java @@ -70,7 +70,7 @@ public ResponseEntity> cdcStatus() { statusBody.put("registeredTargets", targetRegistry.all().stream() .map(targetConnector -> { Map targetEntry = new LinkedHashMap<>(); - targetEntry.put("id", targetConnector.id()); + targetEntry.put("id", targetConnector.targetId()); targetEntry.put("displayName", targetConnector.displayName()); targetEntry.put("scaffoldOnly", targetConnector.scaffoldOnly()); return targetEntry; @@ -93,7 +93,7 @@ public ResponseEntity>> cdcTargets() { List> targetEntries = targetRegistry.all().stream() .map(targetConnector -> { Map targetEntry = new LinkedHashMap<>(); - targetEntry.put("id", targetConnector.id()); + targetEntry.put("id", targetConnector.targetId()); targetEntry.put("displayName", targetConnector.displayName()); targetEntry.put("scaffoldOnly", targetConnector.scaffoldOnly()); return targetEntry; @@ -106,7 +106,7 @@ private Map sourceRegistryEntry( com.xtrmetl.cdc.spi.CdcSourceConnector sourceConnector ) { Map sourceEntry = new LinkedHashMap<>(); - sourceEntry.put("id", sourceConnector.id()); + sourceEntry.put("id", sourceConnector.sourceId()); sourceEntry.put("displayName", sourceConnector.displayName()); sourceEntry.put("engine", sourceConnector.capabilities().engine()); sourceEntry.put("databases", sourceConnector.capabilities().databases()); From 7b54bf426b58f118cb68e008bf466ac6b5cad65f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 02:56:42 +0900 Subject: [PATCH 18/36] test(cdc): use semantic registry identities --- .../cdc/spi/CdcRegistryIdentityTest.java | 95 ++++++++++--------- 1 file changed, 52 insertions(+), 43 deletions(-) diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcRegistryIdentityTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcRegistryIdentityTest.java index 70c89562..7a9f0c6e 100644 --- a/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcRegistryIdentityTest.java +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcRegistryIdentityTest.java @@ -25,32 +25,32 @@ class CdcRegistryIdentityTest { @Test void duplicateSourceConnectorIdsFailClosedInsteadOfReplacingRegistration() { - CdcSourceConnector first = source("duplicate-source"); - CdcSourceConnector second = source("duplicate-source"); - CdcSourceRegistry registry = new CdcSourceRegistry(List.of(first)); + CdcSourceConnector firstSourceConnector = sourceConnector("duplicate-source"); + CdcSourceConnector secondSourceConnector = sourceConnector("duplicate-source"); + CdcSourceRegistry sourceRegistry = new CdcSourceRegistry(List.of(firstSourceConnector)); - IllegalArgumentException failure = assertThrows( + IllegalArgumentException registrationFailure = assertThrows( IllegalArgumentException.class, - () -> registry.register(second) + () -> sourceRegistry.register(secondSourceConnector) ); - assertEquals("Duplicate CDC source connector id: duplicate-source", failure.getMessage()); - assertSame(first, registry.find("duplicate-source").orElseThrow()); + assertEquals("Duplicate CDC source connector id: duplicate-source", registrationFailure.getMessage()); + assertSame(firstSourceConnector, sourceRegistry.find("duplicate-source").orElseThrow()); } @Test void duplicateTargetConnectorIdsFailClosedInsteadOfReplacingRegistration() { - CdcTargetRegistry registry = new CdcTargetRegistry(); - CdcTargetConnector originalKafka = registry.find(KafkaCdcTargetConnector.ID).orElseThrow(); - CdcTargetConnector duplicateKafka = target(KafkaCdcTargetConnector.ID); + CdcTargetRegistry targetRegistry = new CdcTargetRegistry(); + CdcTargetConnector originalKafkaTarget = targetRegistry.find(KafkaCdcTargetConnector.TARGET_ID).orElseThrow(); + CdcTargetConnector duplicateKafkaTarget = targetConnector(KafkaCdcTargetConnector.TARGET_ID); - IllegalArgumentException failure = assertThrows( + IllegalArgumentException registrationFailure = assertThrows( IllegalArgumentException.class, - () -> registry.register(duplicateKafka) + () -> targetRegistry.register(duplicateKafkaTarget) ); - assertEquals("Duplicate CDC target connector id: kafka", failure.getMessage()); - assertSame(originalKafka, registry.find(KafkaCdcTargetConnector.ID).orElseThrow()); + assertEquals("Duplicate CDC target connector id: kafka", registrationFailure.getMessage()); + assertSame(originalKafkaTarget, targetRegistry.find(KafkaCdcTargetConnector.TARGET_ID).orElseThrow()); } @Test @@ -69,60 +69,69 @@ void springDiscoveryConstructorIsExplicitlyAutowired() { @Test void nullSourceConnectorFailsBeforeRegistryMutation() { - CdcSourceRegistry registry = new CdcSourceRegistry(); - IllegalArgumentException failure = assertThrows(IllegalArgumentException.class, () -> registry.register(null)); - assertEquals("CDC source connector must not be null", failure.getMessage()); + CdcSourceRegistry sourceRegistry = new CdcSourceRegistry(); + IllegalArgumentException registrationFailure = assertThrows( + IllegalArgumentException.class, + () -> sourceRegistry.register(null) + ); + assertEquals("CDC source connector must not be null", registrationFailure.getMessage()); } @Test void nullTargetConnectorFailsBeforeRegistryMutation() { - CdcTargetRegistry registry = new CdcTargetRegistry(); - IllegalArgumentException failure = assertThrows(IllegalArgumentException.class, () -> registry.register(null)); - assertEquals("CDC target connector must not be null", failure.getMessage()); + CdcTargetRegistry targetRegistry = new CdcTargetRegistry(); + IllegalArgumentException registrationFailure = assertThrows( + IllegalArgumentException.class, + () -> targetRegistry.register(null) + ); + assertEquals("CDC target connector must not be null", registrationFailure.getMessage()); } @Test void blankSourceConnectorIdFailsBeforeRegistryMutation() { - CdcSourceConnector blank = source(" "); - IllegalArgumentException failure = assertThrows( + CdcSourceConnector blankSourceConnector = sourceConnector(" "); + IllegalArgumentException registrationFailure = assertThrows( IllegalArgumentException.class, - () -> new CdcSourceRegistry(List.of(blank)) + () -> new CdcSourceRegistry(List.of(blankSourceConnector)) ); - assertEquals("CDC source connector id must not be blank", failure.getMessage()); + assertEquals("CDC source connector id must not be blank", registrationFailure.getMessage()); } @Test void blankTargetConnectorIdFailsBeforeRegistryMutation() { - CdcTargetRegistry registry = new CdcTargetRegistry(); - CdcTargetConnector blank = target(""); - IllegalArgumentException failure = assertThrows(IllegalArgumentException.class, () -> registry.register(blank)); - assertEquals("CDC target connector id must not be blank", failure.getMessage()); + CdcTargetRegistry targetRegistry = new CdcTargetRegistry(); + CdcTargetConnector blankTargetConnector = targetConnector(""); + IllegalArgumentException registrationFailure = assertThrows( + IllegalArgumentException.class, + () -> targetRegistry.register(blankTargetConnector) + ); + assertEquals("CDC target connector id must not be blank", registrationFailure.getMessage()); } @Test void sourceConnectorCollectionCannotDeleteRegistrationAuthority() { - CdcSourceRegistry registry = new CdcSourceRegistry(List.of(source("immutable-source"))); - assertThrows(UnsupportedOperationException.class, () -> registry.all().clear()); - assertTrue(registry.find("immutable-source").isPresent()); + CdcSourceRegistry sourceRegistry = new CdcSourceRegistry(List.of(sourceConnector("immutable-source"))); + assertThrows(UnsupportedOperationException.class, () -> sourceRegistry.all().clear()); + assertTrue(sourceRegistry.find("immutable-source").isPresent()); } @Test void targetConnectorCollectionCannotDeleteRegistrationAuthority() { - CdcTargetRegistry registry = new CdcTargetRegistry(); - assertThrows(UnsupportedOperationException.class, () -> registry.all().clear()); - assertTrue(registry.find(KafkaCdcTargetConnector.ID).isPresent()); - assertTrue(registry.find(JdbcReplicaCdcTargetConnector.ID).isPresent()); + CdcTargetRegistry targetRegistry = new CdcTargetRegistry(); + assertThrows(UnsupportedOperationException.class, () -> targetRegistry.all().clear()); + assertTrue(targetRegistry.find(KafkaCdcTargetConnector.TARGET_ID).isPresent()); + assertTrue(targetRegistry.find(JdbcReplicaCdcTargetConnector.TARGET_ID).isPresent()); } - private static CdcSourceConnector source(String id) { - CdcSourceConnector connector = mock(CdcSourceConnector.class); - when(connector.id()).thenReturn(id); - return connector; + private static CdcSourceConnector sourceConnector(String sourceId) { + CdcSourceConnector sourceConnector = mock(CdcSourceConnector.class); + when(sourceConnector.sourceId()).thenReturn(sourceId); + return sourceConnector; } - private static CdcTargetConnector target(String id) { - CdcTargetConnector connector = mock(CdcTargetConnector.class); - when(connector.id()).thenReturn(id); - return connector; + private static CdcTargetConnector targetConnector(String targetId) { + CdcTargetConnector targetConnector = mock(CdcTargetConnector.class); + when(targetConnector.targetId()).thenReturn(targetId); + return targetConnector; } } From 5627d5598eb5558e53b1516a6a7fc3560c2578e0 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 02:57:04 +0900 Subject: [PATCH 19/36] test(cdc): wire source registry by semantic id --- .../CdcSourceRegistrySpringWiringTest.java | 29 ++++++++++++------- 1 file changed, 18 insertions(+), 11 deletions(-) diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcSourceRegistrySpringWiringTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcSourceRegistrySpringWiringTest.java index cb0e4f7d..4b5ab40b 100644 --- a/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcSourceRegistrySpringWiringTest.java +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcSourceRegistrySpringWiringTest.java @@ -19,18 +19,18 @@ class CdcSourceRegistrySpringWiringTest { @Test void springContextRegistersDiscoveredSourceConnectorBean() { - TestSourceConnector connector = new TestSourceConnector(); + TestSourceConnector sourceConnector = new TestSourceConnector(); - try (AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext()) { - context.registerBean(CdcSourceConnector.class, () -> connector); - context.register(CdcSourceRegistry.class); - context.refresh(); + try (AnnotationConfigApplicationContext applicationContext = new AnnotationConfigApplicationContext()) { + applicationContext.registerBean(CdcSourceConnector.class, () -> sourceConnector); + applicationContext.register(CdcSourceRegistry.class); + applicationContext.refresh(); - CdcSourceRegistry registry = context.getBean(CdcSourceRegistry.class); + CdcSourceRegistry sourceRegistry = applicationContext.getBean(CdcSourceRegistry.class); assertSame( - connector, - registry.find(connector.id()).orElseThrow(), + sourceConnector, + sourceRegistry.find(sourceConnector.sourceId()).orElseThrow(), "Spring must construct the registry through its connector-provider constructor" ); } @@ -39,10 +39,17 @@ void springContextRegistersDiscoveredSourceConnectorBean() { private static final class TestSourceConnector implements CdcSourceConnector { @Override - public String id() { + public String sourceId() { return "test_source"; } + /** @deprecated compatibility fixture for the historical SPI accessor. */ + @Override + @Deprecated(forRemoval = false) + public String id() { + return sourceId(); + } + @Override public String displayName() { return "Test source"; @@ -54,12 +61,12 @@ public SourceCapabilities capabilities() { } @Override - public void validate(Map config) { + public void validate(Map sourceConfig) { // No configuration is required for this constructor-selection regression fixture. } @Override - public void start(Map config) { + public void start(Map sourceConfig) { // No runtime capture is required for this constructor-selection regression fixture. } From be23695f6d0569e59e8001c041d38332c4758d5c Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 02:57:32 +0900 Subject: [PATCH 20/36] test(cdc): pin legacy identity adapters --- .../spi/CdcConnectorSemanticIdentityTest.java | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcConnectorSemanticIdentityTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcConnectorSemanticIdentityTest.java index 1e1c69b3..75ee49b2 100644 --- a/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcConnectorSemanticIdentityTest.java +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcConnectorSemanticIdentityTest.java @@ -10,18 +10,26 @@ class CdcConnectorSemanticIdentityTest { @Test - void sourceConnectorExposesSourceIdentityBySemanticName() { + @SuppressWarnings("deprecation") + void sourceConnectorExposesSemanticIdentityAndPreservesLegacyAlias() { CdcSourceConnector sourceConnector = new PostgresDebeziumCdcSource(); - assertEquals("postgres-debezium", sourceConnector.sourceId()); + assertEquals(PostgresDebeziumCdcSource.SOURCE_ID, sourceConnector.sourceId()); + assertEquals(sourceConnector.sourceId(), sourceConnector.id()); + assertEquals(PostgresDebeziumCdcSource.SOURCE_ID, PostgresDebeziumCdcSource.ID); } @Test - void targetConnectorsExposeTargetIdentityBySemanticName() { + @SuppressWarnings("deprecation") + void targetConnectorsExposeSemanticIdentityAndPreserveLegacyAliases() { CdcTargetConnector kafkaTargetConnector = new KafkaCdcTargetConnector(); CdcTargetConnector jdbcTargetConnector = new JdbcReplicaCdcTargetConnector(); - assertEquals("kafka", kafkaTargetConnector.targetId()); - assertEquals("jdbc-replica", jdbcTargetConnector.targetId()); + assertEquals(KafkaCdcTargetConnector.TARGET_ID, kafkaTargetConnector.targetId()); + assertEquals(JdbcReplicaCdcTargetConnector.TARGET_ID, jdbcTargetConnector.targetId()); + assertEquals(kafkaTargetConnector.targetId(), kafkaTargetConnector.id()); + assertEquals(jdbcTargetConnector.targetId(), jdbcTargetConnector.id()); + assertEquals(KafkaCdcTargetConnector.TARGET_ID, KafkaCdcTargetConnector.ID); + assertEquals(JdbcReplicaCdcTargetConnector.TARGET_ID, JdbcReplicaCdcTargetConnector.ID); } } From c1d6f0cbd03ff1eb3da3a5ce58d7780a2bd7cbab Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 02:58:23 +0900 Subject: [PATCH 21/36] test(cdc): use semantic target constants --- .../xtrmetl/cdc/spi/CdcTargetRegistryTest.java | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcTargetRegistryTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcTargetRegistryTest.java index 55d84c4c..f5e82f84 100644 --- a/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcTargetRegistryTest.java +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcTargetRegistryTest.java @@ -14,15 +14,15 @@ class CdcTargetRegistryTest { @Test void registersKafkaAndJdbcReplicaTargets() { - CdcTargetRegistry registry = new CdcTargetRegistry(); + CdcTargetRegistry targetRegistry = new CdcTargetRegistry(); - assertEquals(2, registry.all().size()); - assertTrue(registry.find(KafkaCdcTargetConnector.ID).isPresent()); - assertTrue(registry.find(JdbcReplicaCdcTargetConnector.ID).isPresent()); + assertEquals(2, targetRegistry.all().size()); + assertTrue(targetRegistry.find(KafkaCdcTargetConnector.TARGET_ID).isPresent()); + assertTrue(targetRegistry.find(JdbcReplicaCdcTargetConnector.TARGET_ID).isPresent()); - CdcTargetConnector kafka = registry.find(KafkaCdcTargetConnector.ID).orElseThrow(); - assertFalse(kafka.scaffoldOnly()); - kafka.validate(Map.of()); - assertThrows(UnsupportedOperationException.class, () -> kafka.write(List.of())); + CdcTargetConnector kafkaTargetConnector = targetRegistry.find(KafkaCdcTargetConnector.TARGET_ID).orElseThrow(); + assertFalse(kafkaTargetConnector.scaffoldOnly()); + kafkaTargetConnector.validate(Map.of()); + assertThrows(UnsupportedOperationException.class, () -> kafkaTargetConnector.write(List.of())); } } From 66f8497ba1194679b0782c198aec41b64874745a Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 02:58:52 +0900 Subject: [PATCH 22/36] refactor(cdc): name MySQL source identity explicitly --- .../com/xtrmetl/cdc/spi/MysqlDebeziumCdcSource.java | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/MysqlDebeziumCdcSource.java b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/MysqlDebeziumCdcSource.java index f3624891..d80ccf60 100644 --- a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/MysqlDebeziumCdcSource.java +++ b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/MysqlDebeziumCdcSource.java @@ -10,9 +10,15 @@ @Component public final class MysqlDebeziumCdcSource extends AbstractScaffoldCdcSource { - public static final String ID = "mysql-debezium"; + public static final String SOURCE_ID = "mysql-debezium"; + + /** + * @deprecated compatibility alias; organization-owned callers use {@link #SOURCE_ID} + */ + @Deprecated(forRemoval = false) + public static final String ID = SOURCE_ID; public MysqlDebeziumCdcSource() { - super(ID, "MySQL (Debezium scaffold)", "debezium-embedded", Set.of("mysql")); + super(SOURCE_ID, "MySQL (Debezium scaffold)", "debezium-embedded", Set.of("mysql")); } } From ef70cd0a5bd9aff7bf9359687d2e9a0f88fcdb92 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 02:59:07 +0900 Subject: [PATCH 23/36] refactor(cdc): name SQL Server source identity explicitly --- .../xtrmetl/cdc/spi/SqlServerDebeziumCdcSource.java | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/SqlServerDebeziumCdcSource.java b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/SqlServerDebeziumCdcSource.java index 29eed3aa..fb14fdad 100644 --- a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/SqlServerDebeziumCdcSource.java +++ b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/SqlServerDebeziumCdcSource.java @@ -10,9 +10,15 @@ @Component public final class SqlServerDebeziumCdcSource extends AbstractScaffoldCdcSource { - public static final String ID = "sqlserver-debezium"; + public static final String SOURCE_ID = "sqlserver-debezium"; + + /** + * @deprecated compatibility alias; organization-owned callers use {@link #SOURCE_ID} + */ + @Deprecated(forRemoval = false) + public static final String ID = SOURCE_ID; public SqlServerDebeziumCdcSource() { - super(ID, "SQL Server (Debezium scaffold)", "debezium-embedded", Set.of("sqlserver")); + super(SOURCE_ID, "SQL Server (Debezium scaffold)", "debezium-embedded", Set.of("sqlserver")); } } From f5bc0f737efbd6f02ec4a4ee9a8e4e35d0f35171 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 02:59:32 +0900 Subject: [PATCH 24/36] test(cdc): use semantic source constants --- .../cdc/spi/CdcSourceRegistryTest.java | 46 +++++++++---------- 1 file changed, 23 insertions(+), 23 deletions(-) diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcSourceRegistryTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcSourceRegistryTest.java index 020f6329..f0323e15 100644 --- a/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcSourceRegistryTest.java +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcSourceRegistryTest.java @@ -20,62 +20,62 @@ class CdcSourceRegistryTest { @Test void defaultConstructorRegistersPostgresOnly() { - CdcSourceRegistry registry = new CdcSourceRegistry(); + CdcSourceRegistry sourceRegistry = new CdcSourceRegistry(); - assertEquals(1, registry.all().size()); - CdcSourceConnector source = registry.find(PostgresDebeziumCdcSource.ID).orElseThrow(); - assertEquals("PostgreSQL (Debezium embedded)", source.displayName()); - assertFalse(source.capabilities().scaffoldOnly()); - assertTrue(source.capabilities().databases().contains("postgresql")); + assertEquals(1, sourceRegistry.all().size()); + CdcSourceConnector postgresSource = sourceRegistry.find(PostgresDebeziumCdcSource.SOURCE_ID).orElseThrow(); + assertEquals("PostgreSQL (Debezium embedded)", postgresSource.displayName()); + assertFalse(postgresSource.capabilities().scaffoldOnly()); + assertTrue(postgresSource.capabilities().databases().contains("postgresql")); } @Test void postgresSpiStartAndStopDelegateToCdcService() throws Exception { - CdcService service = mock(CdcService.class); - ObjectProvider provider = providerFor(service); - PostgresDebeziumCdcSource source = new PostgresDebeziumCdcSource(provider); + CdcService cdcService = mock(CdcService.class); + ObjectProvider cdcServiceProvider = providerFor(cdcService); + PostgresDebeziumCdcSource postgresSource = new PostgresDebeziumCdcSource(cdcServiceProvider); - source.validate(Map.of()); - source.start(Map.of()); - source.stop(); + postgresSource.validate(Map.of()); + postgresSource.start(Map.of()); + postgresSource.stop(); - verify(service).start(); - verify(service).stop(); + verify(cdcService).start(); + verify(cdcService).stop(); } @Test void registersScaffoldSourcesWhenProvided() { - CdcSourceRegistry registry = new CdcSourceRegistry(List.of( + CdcSourceRegistry sourceRegistry = new CdcSourceRegistry(List.of( new PostgresDebeziumCdcSource(), new MysqlDebeziumCdcSource(), new SqlServerDebeziumCdcSource() )); - assertEquals(3, registry.all().size()); - assertTrue(registry.find(MysqlDebeziumCdcSource.ID).orElseThrow().capabilities().scaffoldOnly()); - assertTrue(registry.find(SqlServerDebeziumCdcSource.ID).orElseThrow().capabilities().scaffoldOnly()); + assertEquals(3, sourceRegistry.all().size()); + assertTrue(sourceRegistry.find(MysqlDebeziumCdcSource.SOURCE_ID).orElseThrow().capabilities().scaffoldOnly()); + assertTrue(sourceRegistry.find(SqlServerDebeziumCdcSource.SOURCE_ID).orElseThrow().capabilities().scaffoldOnly()); } - private static ObjectProvider providerFor(CdcService service) { + private static ObjectProvider providerFor(CdcService cdcService) { return new ObjectProvider<>() { @Override public CdcService getObject() { - return service; + return cdcService; } @Override public CdcService getObject(Object... args) { - return service; + return cdcService; } @Override public CdcService getIfAvailable() { - return service; + return cdcService; } @Override public CdcService getIfUnique() { - return service; + return cdcService; } }; } From fd1e91e801a8ef1ab9c8cd8ee350c95749bfa732 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 03:01:00 +0900 Subject: [PATCH 25/36] refactor(cdc): use semantic PostgreSQL source constant --- .../src/main/java/com/xtrmetl/cdc/service/CdcService.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/service/CdcService.java b/cdc-service/src/main/java/com/xtrmetl/cdc/service/CdcService.java index d33b6525..39493ea0 100644 --- a/cdc-service/src/main/java/com/xtrmetl/cdc/service/CdcService.java +++ b/cdc-service/src/main/java/com/xtrmetl/cdc/service/CdcService.java @@ -269,7 +269,7 @@ private void maybeMapCanonical(String topic, String key, String value) { } try { boolean ok = changeRecordMapper - .map(PostgresDebeziumCdcSource.ID, topic, key, value) + .map(PostgresDebeziumCdcSource.SOURCE_ID, topic, key, value) .isPresent(); if (ok) { canonicalMapSuccess.incrementAndGet(); From dcc9694257f87bc5b3c5804bdc647ac9426ca6bd Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 03:01:56 +0900 Subject: [PATCH 26/36] fix(cdc): preserve source validation messages --- .../java/com/xtrmetl/cdc/spi/PostgresDebeziumCdcSource.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/PostgresDebeziumCdcSource.java b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/PostgresDebeziumCdcSource.java index 67173d9c..297699dc 100644 --- a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/PostgresDebeziumCdcSource.java +++ b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/PostgresDebeziumCdcSource.java @@ -103,11 +103,11 @@ public SourceCapabilities capabilities() { @Override public void validate(Map sourceConfig) { if (sourceConfig == null) { - throw new IllegalArgumentException("sourceConfig must not be null"); + throw new IllegalArgumentException("config must not be null"); } if (!sourceConfig.isEmpty()) { throw new IllegalArgumentException( - "postgres-debezium uses deployment-owned configuration; per-call sourceConfig must be empty" + "postgres-debezium uses deployment-owned configuration; per-call config must be empty" ); } } From 053550a8b563d4b269d9bc387756ae5d7c5a729a Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 03:02:15 +0900 Subject: [PATCH 27/36] fix(cdc): preserve Kafka validation message --- .../main/java/com/xtrmetl/cdc/spi/KafkaCdcTargetConnector.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/KafkaCdcTargetConnector.java b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/KafkaCdcTargetConnector.java index 2e38a33a..f9e7cc9e 100644 --- a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/KafkaCdcTargetConnector.java +++ b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/KafkaCdcTargetConnector.java @@ -54,7 +54,7 @@ public Capabilities capabilities() { @Override public void validate(Map targetConfig) { if (targetConfig == null) { - throw new IllegalArgumentException("targetConfig must not be null"); + throw new IllegalArgumentException("config must not be null"); } } From 2417ed91d5e7bcbcfb40b649aa5bbd9f7791b00b Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 03:02:52 +0900 Subject: [PATCH 28/36] fix(cdc): preserve JDBC validation message --- .../java/com/xtrmetl/cdc/spi/JdbcReplicaCdcTargetConnector.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/JdbcReplicaCdcTargetConnector.java b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/JdbcReplicaCdcTargetConnector.java index 96055e2c..7a9f068a 100644 --- a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/JdbcReplicaCdcTargetConnector.java +++ b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/JdbcReplicaCdcTargetConnector.java @@ -53,7 +53,7 @@ public Capabilities capabilities() { @Override public void validate(Map targetConfig) { if (targetConfig == null) { - throw new IllegalArgumentException("targetConfig must not be null"); + throw new IllegalArgumentException("config must not be null"); } } From 1624540064be47a72a49be5f5de0f4bf0abb52e0 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 03:03:50 +0900 Subject: [PATCH 29/36] docs(cdc): document semantic connector identity boundary --- .../cdc-source-semantic-identifiers.md | 25 +++++++++++-------- 1 file changed, 15 insertions(+), 10 deletions(-) diff --git a/docs/doctoring/cdc-source-semantic-identifiers.md b/docs/doctoring/cdc-source-semantic-identifiers.md index fde67b23..d7de47ef 100644 --- a/docs/doctoring/cdc-source-semantic-identifiers.md +++ b/docs/doctoring/cdc-source-semantic-identifiers.md @@ -2,38 +2,43 @@ ## Decision -The CDC source configuration bounded context owns source identity and connector-type vocabulary. Internal Java identifiers therefore use the explicit terms `sourceId` and `sourceType` rather than the generic one-word `id` and `type` names. The Java casing remains idiomatic camelCase; the decision is about semantic specificity, not forcing snake_case into Java. +The CDC configuration and connector-discovery bounded context owns source identity, target identity, and connector-type vocabulary. Internal Java identifiers therefore use explicit terms such as `sourceId`, `targetId`, and `sourceType` rather than generic one-word `id` and `type` names. Java casing remains idiomatic camelCase; the decision is about semantic specificity, not forcing snake_case into Java. `CdcSourceFactory.SourceSpec` is the authoritative internal value object for one declared source. Its record components are `sourceId`, `sourceType`, and `enabled`. Factory parameters and local variables use source-specific terms (`sourceRegistry`, `sourceSpecs`, `sourceDescription`, `sourceConnector`) so ownership remains visible at the call site. +The connector SPIs now expose semantic organization-owned accessors as well. `CdcSourceConnector.sourceId()` is the internal source-identity vocabulary and `CdcTargetConnector.targetId()` is the internal target-identity vocabulary. Built-in connectors publish semantic constants (`SOURCE_ID` or `TARGET_ID`), registries index by semantic identifiers, controller code reads the semantic accessors, and internal tests use the same language. + ## Compatibility boundary -Existing operators already configure `xtrmetl.cdc.sources[*].id` and `xtrmetl.cdc.sources[*].type`, and the CDC status resource already emits configured-source objects containing `id` and `type`. This change therefore treats those historical keys as an anti-corruption/compatibility boundary rather than silently changing the wire/config contract. +Existing operators already configure `xtrmetl.cdc.sources[*].id` and `xtrmetl.cdc.sources[*].type`, and the CDC status resources already emit source/target objects containing `id`. This repair treats those historical keys as anti-corruption/compatibility boundaries rather than silently changing wire/config contracts. `XtrmetlProperties.Source` stores values internally as `sourceId` and `sourceType`. The historical `getId`/`setId` and `getType`/`setType` JavaBean accessors remain only so Spring can continue binding the established configuration keys. Repository-owned production callers use `getSourceId`/`getSourceType`. Focused binder coverage proves that legacy `id`/`type` configuration still populates the semantic internal fields. -`CdcSourceFactory.describeConfigured` likewise keeps the established output keys `id` and `type`, while all internal record access is through `sourceId()` and `sourceType()`. The compatibility regression explicitly fails if the implementation starts leaking new Java member names into the existing HTTP payload. +`CdcSourceFactory.describeConfigured` likewise keeps the established output keys `id` and `type`, while all internal record access is through `sourceId()` and `sourceType()`. `CdcController` keeps the established response key `id`, but obtains its value from `sourceId()` or `targetId()` internally. The compatibility regression explicitly fails if implementation member names leak into the existing HTTP payload. + +The CDC connector SPIs can also be consumed outside the repository. Their historical generic `id()` accessor therefore remains deprecated as a compatibility seam while organization-owned callers use `sourceId()`/`targetId()`. Built-in connectors retain deprecated `ID` constants as aliases to `SOURCE_ID`/`TARGET_ID`; repository-owned callers use the semantic constants. This keeps old caller source behavior available without making the generic vocabulary authoritative inside the bounded context. ## DDD traceability -- **Bounded context:** Change Data Capture source configuration and discovery. -- **Ubiquitous language:** CDC source, source identifier, source connector type, source registry, configured source, registered source. +- **Bounded context:** Change Data Capture source configuration, source discovery, and target discovery. +- **Ubiquitous language:** CDC source, source identifier, source connector type, source registry, CDC target, target identifier, target registry, configured source, registered source, registered target. - **Value object:** `CdcSourceFactory.SourceSpec` represents one immutable source declaration. -- **Domain service:** `CdcSourceFactory` validates source identity uniqueness and resolves source connector types. +- **Domain services:** `CdcSourceFactory`, `CdcSourceRegistry`, and `CdcTargetRegistry` validate and resolve semantic connector identities. - **Invariant:** two configured source declarations cannot share the same `sourceId`. +- **Invariant:** registered source connectors cannot share a `sourceId`; registered target connectors cannot share a `targetId`. - **Invariant:** `sourceId` and `sourceType` must be nonblank before a declaration enters the configured-source description pipeline. -- **Compatibility invariant:** established operator configuration and HTTP keys remain `id`/`type` until a separately versioned breaking contract intentionally replaces them. +- **Compatibility invariant:** established operator configuration and HTTP keys remain `id`/`type`, and the deprecated SPI aliases return the same identity as the semantic accessors, until a separately versioned breaking contract intentionally replaces them. - **Persistence:** no database row, column, index, migration, Kafka record, or Debezium offset changes in this repair. ## Verification contract -Focused tests require semantic record accessors, duplicate-source rejection, repeated connector types with distinct source identities, unchanged legacy HTTP keys, semantic Java defaults, and Spring binding of the historical `id`/`type` keys into the new internal fields. The full Maven/CI, dependency review, SBOM, SAST, and Security Scan gates remain authoritative on the unchanged resulting head. +Focused tests require semantic record and connector accessors, duplicate source/target rejection, repeated connector types with distinct source identities, unchanged legacy HTTP keys, semantic Java defaults/constants, compatibility alias equality, and Spring binding of the historical `id`/`type` keys into the new internal fields. Existing validation exception text is preserved so a naming-only refactor does not create an unrelated behavioral contract change. The full Maven/CI, dependency review, SBOM, SAST, and Security Scan gates remain authoritative on the unchanged resulting head. -The regression-only head was intentionally superseded quickly by ordinary non-force implementation commits, so its hosted jobs were cancelled rather than claimed as RED evidence. The regression itself is retained in history and remains executable on the final branch; no cancelled or predecessor run is counted as merge evidence. +The regression-only heads were intentionally superseded quickly by ordinary non-force implementation commits. The connector-identity RED commit introduced calls to `sourceId()`/`targetId()` before production support landed; successor commits implement the contract while retaining explicit compatibility adapters. No cancelled, predecessor, base-only, or model-only run is counted as merge evidence. ## Research basis -The naming rule is intentionally semantic rather than mechanical. Schankin et al. found that descriptive compound identifiers helped experienced Java developers locate semantic defects faster than shorter, less descriptive names. Feitelson et al. model naming as selecting the concepts a name should communicate, choosing words for those concepts, and constructing the identifier; explicitly applying that model produced names judged superior to unconstrained choices. These findings support encoding the owned concepts (`source` + `id`, `source` + `type`) while retaining the host language's normal casing and allowing concise names where context itself already supplies the meaning. +The naming rule is intentionally semantic rather than mechanical. Schankin et al. found that descriptive compound identifiers helped experienced Java developers locate semantic defects faster than shorter, less descriptive names. Feitelson et al. model naming as selecting the concepts a name should communicate, choosing words for those concepts, and constructing the identifier; explicitly applying that model produced names judged superior to unconstrained choices. These findings support encoding the owned concepts (`source` + `id`, `target` + `id`, `source` + `type`) while retaining the host language's normal casing and allowing concise names where context itself already supplies the meaning. ### References From 4a059410a8facf1b1dd0027a75723bf417f3186e Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 03:05:37 +0900 Subject: [PATCH 30/36] test(etl): require semantic target identity --- .../TargetConnectorSemanticIdentityTest.java | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) create mode 100644 etl-service/src/test/java/com/xtrmetl/etl/connector/TargetConnectorSemanticIdentityTest.java diff --git a/etl-service/src/test/java/com/xtrmetl/etl/connector/TargetConnectorSemanticIdentityTest.java b/etl-service/src/test/java/com/xtrmetl/etl/connector/TargetConnectorSemanticIdentityTest.java new file mode 100644 index 00000000..f225ece1 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/connector/TargetConnectorSemanticIdentityTest.java @@ -0,0 +1,18 @@ +package com.xtrmetl.etl.connector; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * Guards bounded-context-specific target identity on the organization-owned ETL connector SPI. + */ +class TargetConnectorSemanticIdentityTest { + + @Test + void targetConnectorExposesTargetIdentityBySemanticName() { + TargetConnector databricksTargetConnector = new DatabricksTargetConnector(); + + assertEquals("databricks", databricksTargetConnector.targetId()); + } +} From 64751339bead26c97e54d9391b6c746dd22eb29c Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 03:06:26 +0900 Subject: [PATCH 31/36] refactor(etl): expose semantic target identity --- .../etl/connector/TargetConnector.java | 28 ++++++++++++++++--- 1 file changed, 24 insertions(+), 4 deletions(-) diff --git a/etl-service/src/main/java/com/xtrmetl/etl/connector/TargetConnector.java b/etl-service/src/main/java/com/xtrmetl/etl/connector/TargetConnector.java index bc51b1f9..a1a461f9 100644 --- a/etl-service/src/main/java/com/xtrmetl/etl/connector/TargetConnector.java +++ b/etl-service/src/main/java/com/xtrmetl/etl/connector/TargetConnector.java @@ -11,6 +11,26 @@ */ public interface TargetConnector extends AutoCloseable { + /** + * Returns the bounded-context-specific ETL target identifier. + * + *

Organization-owned callers use this semantic accessor. The historical + * {@link #id()} method remains only as a compatibility seam for existing connector + * implementations and callers.

+ * + * @return exact ETL target identifier + */ + default String targetId() { + return id(); + } + + /** + * Historical compatibility accessor for the generic connector identifier. + * + * @return exact ETL target identifier + * @deprecated organization-owned callers must use {@link #targetId()} + */ + @Deprecated(forRemoval = false) String id(); String displayName(); @@ -55,20 +75,20 @@ default Map describeIntegration() { /** * Validate configuration before open. Implementations should fail fast on missing secrets. */ - void validate(Map config); + void validate(Map targetConfig); /** * Establish client resources (connections, tokens). No-op allowed for pure scaffolds. */ - default void open(Map config) { - validate(config); + default void open(Map targetConfig) { + validate(targetConfig); } /** * Write a batch of change records. Scaffold implementations must throw * {@link UnsupportedOperationException}. */ - void write(List batch); + void write(List changeBatch); @Override default void close() { From 6aa7568d697e3fedfcca5b2991520b2e9638b198 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 03:08:15 +0900 Subject: [PATCH 32/36] refactor(etl): make scaffold target names semantic --- .../AbstractScaffoldTargetConnector.java | 57 +++++++++++-------- 1 file changed, 33 insertions(+), 24 deletions(-) diff --git a/etl-service/src/main/java/com/xtrmetl/etl/connector/AbstractScaffoldTargetConnector.java b/etl-service/src/main/java/com/xtrmetl/etl/connector/AbstractScaffoldTargetConnector.java index 56a2d1fd..296c27fd 100644 --- a/etl-service/src/main/java/com/xtrmetl/etl/connector/AbstractScaffoldTargetConnector.java +++ b/etl-service/src/main/java/com/xtrmetl/etl/connector/AbstractScaffoldTargetConnector.java @@ -12,29 +12,38 @@ */ public abstract class AbstractScaffoldTargetConnector implements TargetConnector { - private final String id; + private final String targetId; private final String displayName; private final List requiredConfigKeys; private final List optionalConfigKeys; - private final Map integration; + private final Map integrationMetadata; protected AbstractScaffoldTargetConnector( - String id, + String targetId, String displayName, List requiredConfigKeys, List optionalConfigKeys, - Map integration + Map integrationMetadata ) { - this.id = Objects.requireNonNull(id, "id"); + this.targetId = Objects.requireNonNull(targetId, "id"); this.displayName = Objects.requireNonNull(displayName, "displayName"); this.requiredConfigKeys = List.copyOf(Objects.requireNonNull(requiredConfigKeys, "requiredConfigKeys")); this.optionalConfigKeys = List.copyOf(Objects.requireNonNull(optionalConfigKeys, "optionalConfigKeys")); - this.integration = Map.copyOf(Objects.requireNonNull(integration, "integration")); + this.integrationMetadata = Map.copyOf(Objects.requireNonNull(integrationMetadata, "integration")); } @Override + public final String targetId() { + return targetId; + } + + /** + * @deprecated compatibility alias; organization-owned callers use {@link #targetId()} + */ + @Override + @Deprecated(forRemoval = false) public final String id() { - return id; + return targetId(); } @Override @@ -65,25 +74,25 @@ public final String writeRefusalReason() { @Override public final Map describeIntegration() { - Map copy = new LinkedHashMap<>(integration); - copy.putIfAbsent("mode", "scaffold"); - copy.putIfAbsent("networkIo", false); - return Map.copyOf(copy); + Map integrationDescription = new LinkedHashMap<>(integrationMetadata); + integrationDescription.putIfAbsent("mode", "scaffold"); + integrationDescription.putIfAbsent("networkIo", false); + return Map.copyOf(integrationDescription); } @Override - public void validate(Map config) { - Objects.requireNonNull(config, "config"); - List missing = new ArrayList<>(); - for (String key : requiredConfigKeys) { - String value = config.get(key); - if (value == null || value.isBlank()) { - missing.add(key); + public void validate(Map targetConfig) { + Objects.requireNonNull(targetConfig, "config"); + List missingConfigKeys = new ArrayList<>(); + for (String configKey : requiredConfigKeys) { + String configValue = targetConfig.get(configKey); + if (configValue == null || configValue.isBlank()) { + missingConfigKeys.add(configKey); } } - if (!missing.isEmpty()) { + if (!missingConfigKeys.isEmpty()) { throw new IllegalArgumentException( - displayName + " config missing required keys: " + missing + displayName + " config missing required keys: " + missingConfigKeys + " (status=SCAFFOLD; live write still refused after validation)" ); } @@ -93,13 +102,13 @@ public void validate(Map config) { * Scaffold open: validate only — no client, no network. */ @Override - public final void open(Map config) { - validate(config); + public final void open(Map targetConfig) { + validate(targetConfig); } @Override - public final void write(List batch) { - Objects.requireNonNull(batch, "batch"); + public final void write(List changeBatch) { + Objects.requireNonNull(changeBatch, "batch"); throw new UnsupportedOperationException(writeRefusalReason()); } } From 75b6639eebbcdba254c3cbe729fa41e3332f4e36 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 03:08:43 +0900 Subject: [PATCH 33/36] refactor(etl): use semantic target ids in registry --- .../etl/connector/TargetConnectorRegistry.java | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/etl-service/src/main/java/com/xtrmetl/etl/connector/TargetConnectorRegistry.java b/etl-service/src/main/java/com/xtrmetl/etl/connector/TargetConnectorRegistry.java index 4e5390af..67e31676 100644 --- a/etl-service/src/main/java/com/xtrmetl/etl/connector/TargetConnectorRegistry.java +++ b/etl-service/src/main/java/com/xtrmetl/etl/connector/TargetConnectorRegistry.java @@ -14,7 +14,7 @@ @Component public class TargetConnectorRegistry { - private final Map byId = new LinkedHashMap<>(); + private final Map targetConnectorsById = new LinkedHashMap<>(); public TargetConnectorRegistry() { register(new DatabricksTargetConnector()); @@ -22,15 +22,15 @@ public TargetConnectorRegistry() { register(new QlikSenseTargetConnector()); } - public final void register(TargetConnector connector) { - byId.put(connector.id(), connector); + public final void register(TargetConnector targetConnector) { + targetConnectorsById.put(targetConnector.targetId(), targetConnector); } - public Optional find(String id) { - return Optional.ofNullable(byId.get(id)); + public Optional find(String targetId) { + return Optional.ofNullable(targetConnectorsById.get(targetId)); } public Collection all() { - return byId.values(); + return targetConnectorsById.values(); } } From d7eb91c4ac03030a4af299ce07d5f5405cb84222 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 03:09:47 +0900 Subject: [PATCH 34/36] refactor(etl): use semantic target names in dispatcher --- .../connector/TargetConnectorDispatcher.java | 135 +++++++++--------- 1 file changed, 69 insertions(+), 66 deletions(-) diff --git a/etl-service/src/main/java/com/xtrmetl/etl/connector/TargetConnectorDispatcher.java b/etl-service/src/main/java/com/xtrmetl/etl/connector/TargetConnectorDispatcher.java index 29186cf0..9e9517c8 100644 --- a/etl-service/src/main/java/com/xtrmetl/etl/connector/TargetConnectorDispatcher.java +++ b/etl-service/src/main/java/com/xtrmetl/etl/connector/TargetConnectorDispatcher.java @@ -33,35 +33,38 @@ public class TargetConnectorDispatcher { private static final Logger log = LoggerFactory.getLogger(TargetConnectorDispatcher.class); - private final TargetConnectorRegistry registry; - private final ConnectorProperties properties; + private final TargetConnectorRegistry targetRegistry; + private final ConnectorProperties connectorProperties; private final ConcurrentMap openedConnectors = new ConcurrentHashMap<>(); private final ConcurrentMap lifecycleLocks = new ConcurrentHashMap<>(); private final ReentrantReadWriteLock lifecycleGate = new ReentrantReadWriteLock(true); - private boolean closed; + private boolean dispatcherClosed; - public TargetConnectorDispatcher(TargetConnectorRegistry registry, ConnectorProperties properties) { - this.registry = Objects.requireNonNull(registry, "registry must not be null"); - this.properties = Objects.requireNonNull(properties, "properties must not be null"); + public TargetConnectorDispatcher( + TargetConnectorRegistry targetRegistry, + ConnectorProperties connectorProperties + ) { + this.targetRegistry = Objects.requireNonNull(targetRegistry, "registry must not be null"); + this.connectorProperties = Objects.requireNonNull(connectorProperties, "properties must not be null"); } @PostConstruct void logConnectorCatalog() { - for (TargetConnector connector : registry.all()) { - boolean enabled = properties.isEnabled(connector.id()); + for (TargetConnector targetConnector : targetRegistry.all()) { + boolean connectorEnabled = connectorProperties.isEnabled(targetConnector.targetId()); log.info( "Target connector id={} status={} enabled={} requiredKeys={}", - connector.id(), - connector.status(), - enabled, - connector.requiredConfigKeys() + targetConnector.targetId(), + targetConnector.status(), + connectorEnabled, + targetConnector.requiredConfigKeys() ); - if (enabled && connector.status() != ConnectorStatus.SUPPORTED) { + if (connectorEnabled && targetConnector.status() != ConnectorStatus.SUPPORTED) { log.warn( "Connector '{}' is enabled but status={} — write() will be refused. " + "See docs/connectors/", - connector.id(), - connector.status() + targetConnector.targetId(), + targetConnector.status() ); } } @@ -78,26 +81,26 @@ void logConnectorCatalog() { * after shutdown begins.

* * @param connectorId registered connector identifier - * @param batch normalized change records to write - * @throws NullPointerException when {@code connectorId} or {@code batch} is null + * @param changeBatch normalized change records to write + * @throws NullPointerException when {@code connectorId} or {@code changeBatch} is null * @throws IllegalArgumentException when the connector identifier is unknown * @throws IllegalStateException when the connector is disabled or the dispatcher is closed * @throws UnsupportedOperationException when the connector is not production-supported */ - public void dispatch(String connectorId, List batch) { + public void dispatch(String connectorId, List changeBatch) { Objects.requireNonNull(connectorId, "connectorId must not be null"); - Objects.requireNonNull(batch, "batch must not be null"); + Objects.requireNonNull(changeBatch, "batch must not be null"); Lock dispatchLock = lifecycleGate.readLock(); dispatchLock.lock(); try { - if (closed) { + if (dispatcherClosed) { throw new IllegalStateException("Target connector dispatcher is closed"); } - TargetConnector connector = registry.find(connectorId) + TargetConnector targetConnector = targetRegistry.find(connectorId) .orElseThrow(() -> new IllegalArgumentException("Unknown connector: " + connectorId)); - if (!properties.isEnabled(connectorId)) { + if (!connectorProperties.isEnabled(connectorId)) { throw new IllegalStateException( "Connector '" + connectorId + "' is disabled. Set xtrmetl.connectors." + connectorId.replace("-", ".") @@ -105,19 +108,19 @@ public void dispatch(String connectorId, List batch) { ); } - Map config = properties.configMap(connectorId); - if (connector.status() != ConnectorStatus.SUPPORTED) { - connector.validate(config); - throw new UnsupportedOperationException(connector.writeRefusalReason()); + Map targetConfig = connectorProperties.configMap(connectorId); + if (targetConnector.status() != ConnectorStatus.SUPPORTED) { + targetConnector.validate(targetConfig); + throw new UnsupportedOperationException(targetConnector.writeRefusalReason()); } - Object connectorLock = lifecycleLocks.computeIfAbsent( + Object targetLock = lifecycleLocks.computeIfAbsent( connectorId, ignored -> new Object() ); - synchronized (connectorLock) { - ensureOpen(connectorId, connector, config); - connector.write(batch); + synchronized (targetLock) { + ensureOpen(connectorId, targetConnector, targetConfig); + targetConnector.write(changeBatch); } } finally { dispatchLock.unlock(); @@ -133,24 +136,24 @@ public List> catalog() { Lock catalogLock = lifecycleGate.readLock(); catalogLock.lock(); try { - List> rows = new ArrayList<>(); - for (TargetConnector connector : registry.all()) { - Map row = new LinkedHashMap<>(); - row.put("id", connector.id()); - row.put("displayName", connector.displayName()); - row.put("status", connector.status().name()); - row.put("enabled", properties.isEnabled(connector.id())); - row.put("writable", !closed - && connector.status() == ConnectorStatus.SUPPORTED - && properties.isEnabled(connector.id())); - row.put("opened", openedConnectors.get(connector.id()) == connector); - row.put("requiredConfigKeys", connector.requiredConfigKeys()); - row.put("optionalConfigKeys", connector.optionalConfigKeys()); - row.put("writeRefusalReason", connector.writeRefusalReason()); - row.put("integration", connector.describeIntegration()); - rows.add(row); + List> catalogRows = new ArrayList<>(); + for (TargetConnector targetConnector : targetRegistry.all()) { + Map catalogRow = new LinkedHashMap<>(); + catalogRow.put("id", targetConnector.targetId()); + catalogRow.put("displayName", targetConnector.displayName()); + catalogRow.put("status", targetConnector.status().name()); + catalogRow.put("enabled", connectorProperties.isEnabled(targetConnector.targetId())); + catalogRow.put("writable", !dispatcherClosed + && targetConnector.status() == ConnectorStatus.SUPPORTED + && connectorProperties.isEnabled(targetConnector.targetId())); + catalogRow.put("opened", openedConnectors.get(targetConnector.targetId()) == targetConnector); + catalogRow.put("requiredConfigKeys", targetConnector.requiredConfigKeys()); + catalogRow.put("optionalConfigKeys", targetConnector.optionalConfigKeys()); + catalogRow.put("writeRefusalReason", targetConnector.writeRefusalReason()); + catalogRow.put("integration", targetConnector.describeIntegration()); + catalogRows.add(catalogRow); } - return rows; + return catalogRows; } finally { catalogLock.unlock(); } @@ -165,25 +168,25 @@ public List> catalog() { */ private void ensureOpen( String connectorId, - TargetConnector connector, - Map config + TargetConnector targetConnector, + Map targetConfig ) { - TargetConnector active = openedConnectors.get(connectorId); - if (active == connector) { + TargetConnector activeConnector = openedConnectors.get(connectorId); + if (activeConnector == targetConnector) { return; } - if (active != null) { + if (activeConnector != null) { throw new IllegalStateException( "Connector registry entry changed after open: " + connectorId ); } - connector.validate(config); + targetConnector.validate(targetConfig); try { - connector.open(config); + targetConnector.open(targetConfig); } catch (RuntimeException openFailure) { try { - connector.close(); + targetConnector.close(); } catch (RuntimeException cleanupFailure) { openFailure.addSuppressed(cleanupFailure); log.warn("Failed to clean up target connector after open failure id={}", connectorId); @@ -191,7 +194,7 @@ private void ensureOpen( throw openFailure; } - openedConnectors.put(connectorId, connector); + openedConnectors.put(connectorId, targetConnector); log.info("Opened target connector id={}", connectorId); } @@ -207,27 +210,27 @@ void closeOpenedConnectors() { Lock shutdownLock = lifecycleGate.writeLock(); shutdownLock.lock(); try { - if (closed) { + if (dispatcherClosed) { return; } - closed = true; + dispatcherClosed = true; - for (Map.Entry entry + for (Map.Entry connectorEntry : new ArrayList<>(openedConnectors.entrySet())) { - String connectorId = entry.getKey(); - TargetConnector connector = entry.getValue(); - Object connectorLock = lifecycleLocks.computeIfAbsent( + String connectorId = connectorEntry.getKey(); + TargetConnector targetConnector = connectorEntry.getValue(); + Object targetLock = lifecycleLocks.computeIfAbsent( connectorId, ignored -> new Object() ); - synchronized (connectorLock) { - if (!openedConnectors.remove(connectorId, connector)) { + synchronized (targetLock) { + if (!openedConnectors.remove(connectorId, targetConnector)) { continue; } try { - connector.close(); + targetConnector.close(); log.info("Closed target connector id={}", connectorId); - } catch (RuntimeException exception) { + } catch (RuntimeException closeFailure) { log.error("Failed to close target connector id={}", connectorId); } } From cd4bade91e8d0d32cf7c73903893b0968e904912 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 03:10:07 +0900 Subject: [PATCH 35/36] test(etl): pin legacy target identity adapter --- .../etl/connector/TargetConnectorSemanticIdentityTest.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/etl-service/src/test/java/com/xtrmetl/etl/connector/TargetConnectorSemanticIdentityTest.java b/etl-service/src/test/java/com/xtrmetl/etl/connector/TargetConnectorSemanticIdentityTest.java index f225ece1..334e16c0 100644 --- a/etl-service/src/test/java/com/xtrmetl/etl/connector/TargetConnectorSemanticIdentityTest.java +++ b/etl-service/src/test/java/com/xtrmetl/etl/connector/TargetConnectorSemanticIdentityTest.java @@ -10,9 +10,11 @@ class TargetConnectorSemanticIdentityTest { @Test - void targetConnectorExposesTargetIdentityBySemanticName() { + @SuppressWarnings("deprecation") + void targetConnectorExposesSemanticIdentityAndPreservesLegacyAlias() { TargetConnector databricksTargetConnector = new DatabricksTargetConnector(); assertEquals("databricks", databricksTargetConnector.targetId()); + assertEquals(databricksTargetConnector.targetId(), databricksTargetConnector.id()); } } From 5daa6c9b5bb1c4ab23c96ce0c5af7073f6b804c7 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 7 Sep 2026 12:09:13 +0900 Subject: [PATCH 36/36] fix(test): supply CDC binding failure exception --- .../cdc/config/XtrmetlPropertiesSecurityDefaultTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/config/XtrmetlPropertiesSecurityDefaultTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/config/XtrmetlPropertiesSecurityDefaultTest.java index 39473019..27d6cf55 100644 --- a/cdc-service/src/test/java/com/xtrmetl/cdc/config/XtrmetlPropertiesSecurityDefaultTest.java +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/config/XtrmetlPropertiesSecurityDefaultTest.java @@ -51,7 +51,7 @@ void legacySourceKeysBindIntoSemanticJavaIdentifiers() { XtrmetlProperties xtrmetlProperties = new Binder(configurationSource) .bind("xtrmetl", Bindable.of(XtrmetlProperties.class)) - .orElseThrow(); + .orElseThrow(() -> new IllegalStateException("xtrmetl properties binding failed")); XtrmetlProperties.Source sourceConfiguration = xtrmetlProperties.getCdc().getSources().getFirst();