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() { 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..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 @@ -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.targetId()); + 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.targetId()); + 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.sourceId()); + 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() + ); } } } 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(); 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" 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. 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"); } } 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()); } } 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() { 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()); } } 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..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 @@ -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) { + public void validate(Map targetConfig) { + if (targetConfig == null) { throw new IllegalArgumentException("config 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" 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..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 @@ -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) { + public void validate(Map targetConfig) { + if (targetConfig == null) { throw new IllegalArgumentException("config 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" 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")); } } 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..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 @@ -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,15 +97,15 @@ 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) { + public void validate(Map sourceConfig) { + if (sourceConfig == null) { throw new IllegalArgumentException("config 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" ); @@ -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); } } } 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")); } } 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..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 @@ -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(() -> new IllegalStateException("xtrmetl properties binding failed")); + XtrmetlProperties.Source sourceConfiguration = + xtrmetlProperties.getCdc().getSources().getFirst(); + + assertEquals("pg-legacy", sourceConfiguration.getSourceId()); + assertEquals("mysql-debezium", sourceConfiguration.getSourceType()); + assertFalse(sourceConfiguration.isEnabled()); + } } 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")) + )); } } 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..75ee49b2 --- /dev/null +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcConnectorSemanticIdentityTest.java @@ -0,0 +1,35 @@ +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 + @SuppressWarnings("deprecation") + void sourceConnectorExposesSemanticIdentityAndPreservesLegacyAlias() { + CdcSourceConnector sourceConnector = new PostgresDebeziumCdcSource(); + + assertEquals(PostgresDebeziumCdcSource.SOURCE_ID, sourceConnector.sourceId()); + assertEquals(sourceConnector.sourceId(), sourceConnector.id()); + assertEquals(PostgresDebeziumCdcSource.SOURCE_ID, PostgresDebeziumCdcSource.ID); + } + + @Test + @SuppressWarnings("deprecation") + void targetConnectorsExposeSemanticIdentityAndPreserveLegacyAliases() { + CdcTargetConnector kafkaTargetConnector = new KafkaCdcTargetConnector(); + CdcTargetConnector jdbcTargetConnector = new JdbcReplicaCdcTargetConnector(); + + 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); + } +} 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; } } 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(); 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. } 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; } }; } 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())); } } diff --git a/docs/doctoring/cdc-source-semantic-identifiers.md b/docs/doctoring/cdc-source-semantic-identifiers.md new file mode 100644 index 00000000..d7de47ef --- /dev/null +++ b/docs/doctoring/cdc-source-semantic-identifiers.md @@ -0,0 +1,47 @@ +# CDC source semantic identifier doctoring + +## Decision + +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 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()`. `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, 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 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`, 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 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 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`, `target` + `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 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()); } } 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() { 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); } } 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(); } } 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..334e16c0 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/connector/TargetConnectorSemanticIdentityTest.java @@ -0,0 +1,20 @@ +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 + @SuppressWarnings("deprecation") + void targetConnectorExposesSemanticIdentityAndPreservesLegacyAlias() { + TargetConnector databricksTargetConnector = new DatabricksTargetConnector(); + + assertEquals("databricks", databricksTargetConnector.targetId()); + assertEquals(databricksTargetConnector.targetId(), databricksTargetConnector.id()); + } +}