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 fd00e11a..17b3d422 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
@@ -12,12 +12,21 @@
/**
* Registry of CDC source connector types discovered as Spring beans, plus a safe
* fallback for unit tests without a Spring context.
+ *
+ *
Connector IDs are unique authority selectors. Duplicate or blank identities fail closed
+ * rather than letting Spring discovery order silently replace an earlier implementation.
*/
@Component
public class CdcSourceRegistry {
private final Map byId = new LinkedHashMap<>();
+ /**
+ * Creates a source registry from Spring-discovered connectors in their configured order.
+ *
+ * @param connectors Spring provider for source connectors
+ * @throws IllegalArgumentException if a discovered connector has a blank or duplicate ID
+ */
public CdcSourceRegistry(ObjectProvider connectors) {
connectors.orderedStream().forEach(this::register);
if (byId.isEmpty()) {
@@ -27,7 +36,10 @@ public CdcSourceRegistry(ObjectProvider connectors) {
}
/**
- * Explicit list for tests.
+ * Creates a source registry from an explicit connector list, primarily for tests and embedding.
+ *
+ * @param connectors connectors to register; a null list behaves like an empty list
+ * @throws IllegalArgumentException if a connector has a blank or duplicate ID
*/
public CdcSourceRegistry(List connectors) {
if (connectors != null) {
@@ -38,18 +50,47 @@ public CdcSourceRegistry(List connectors) {
}
}
+ /**
+ * Creates a registry containing the PostgreSQL Debezium fallback source.
+ */
public CdcSourceRegistry() {
this(List.of());
}
+ /**
+ * Registers one connector under a unique non-blank ID.
+ *
+ * @param connector connector to register
+ * @throws IllegalArgumentException if the connector is null or its ID is blank or already registered
+ */
public final void register(CdcSourceConnector connector) {
- byId.put(connector.id(), connector);
+ if (connector == null) {
+ throw new IllegalArgumentException("CDC source connector must not be null");
+ }
+ String id = connector.id();
+ if (id == null || id.isBlank()) {
+ throw new IllegalArgumentException("CDC source connector id must not be blank");
+ }
+ if (byId.putIfAbsent(id, connector) != null) {
+ throw new IllegalArgumentException("Duplicate CDC source connector id: " + id);
+ }
}
+ /**
+ * Finds a registered source connector by its exact ID.
+ *
+ * @param id connector ID
+ * @return the registered connector, or empty when the ID is unknown
+ */
public Optional find(String id) {
return Optional.ofNullable(byId.get(id));
}
+ /**
+ * Returns all registered source connectors in deterministic registration order.
+ *
+ * @return registered source connectors
+ */
public Collection all() {
return byId.values();
}
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 5d42b186..9a283d66 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
@@ -9,25 +9,57 @@
/**
* Registry of CDC target types (Kafka, JDBC replica, future warehouses).
+ *
+ * Connector IDs are unique authority selectors. Duplicate or blank identities fail closed
+ * rather than silently replacing the implementation selected by operator configuration.
*/
@Component
public class CdcTargetRegistry {
private final Map byId = new LinkedHashMap<>();
+ /**
+ * Creates the target registry with the built-in Kafka and JDBC-replica descriptors.
+ */
public CdcTargetRegistry() {
register(new KafkaCdcTargetConnector());
register(new JdbcReplicaCdcTargetConnector());
}
+ /**
+ * Registers one target connector under a unique non-blank ID.
+ *
+ * @param connector connector to register
+ * @throws IllegalArgumentException if the connector is null or its ID is blank or already registered
+ */
public final void register(CdcTargetConnector connector) {
- byId.put(connector.id(), connector);
+ if (connector == null) {
+ throw new IllegalArgumentException("CDC target connector must not be null");
+ }
+ String id = connector.id();
+ if (id == null || id.isBlank()) {
+ throw new IllegalArgumentException("CDC target connector id must not be blank");
+ }
+ if (byId.putIfAbsent(id, connector) != null) {
+ throw new IllegalArgumentException("Duplicate CDC target connector id: " + id);
+ }
}
+ /**
+ * Finds a registered target connector by its exact ID.
+ *
+ * @param id connector ID
+ * @return the registered connector, or empty when the ID is unknown
+ */
public Optional find(String id) {
return Optional.ofNullable(byId.get(id));
}
+ /**
+ * Returns all registered target connectors in deterministic registration order.
+ *
+ * @return registered target connectors
+ */
public Collection all() {
return byId.values();
}
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
new file mode 100644
index 00000000..8d6ad0c1
--- /dev/null
+++ b/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcRegistryIdentityTest.java
@@ -0,0 +1,100 @@
+package com.xtrmetl.cdc.spi;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class CdcRegistryIdentityTest {
+
+ @Test
+ void duplicateSourceConnectorIdsFailClosedInsteadOfReplacingRegistration() {
+ CdcSourceConnector first = source("duplicate-source");
+ CdcSourceConnector second = source("duplicate-source");
+
+ IllegalArgumentException failure = assertThrows(
+ IllegalArgumentException.class,
+ () -> new CdcSourceRegistry(List.of(first, second))
+ );
+
+ assertEquals("Duplicate CDC source connector id: duplicate-source", failure.getMessage());
+ }
+
+ @Test
+ void duplicateTargetConnectorIdsFailClosedInsteadOfReplacingRegistration() {
+ CdcTargetRegistry registry = new CdcTargetRegistry();
+ CdcTargetConnector duplicateKafka = target(KafkaCdcTargetConnector.ID);
+
+ IllegalArgumentException failure = assertThrows(
+ IllegalArgumentException.class,
+ () -> registry.register(duplicateKafka)
+ );
+
+ assertEquals("Duplicate CDC target connector id: kafka", failure.getMessage());
+ }
+
+ @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());
+ }
+
+ @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());
+ }
+
+ @Test
+ void blankSourceConnectorIdFailsBeforeRegistryMutation() {
+ CdcSourceConnector blank = source(" ");
+
+ IllegalArgumentException failure = assertThrows(
+ IllegalArgumentException.class,
+ () -> new CdcSourceRegistry(List.of(blank))
+ );
+
+ assertEquals("CDC source connector id must not be blank", failure.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());
+ }
+
+ private static CdcSourceConnector source(String id) {
+ CdcSourceConnector connector = mock(CdcSourceConnector.class);
+ when(connector.id()).thenReturn(id);
+ return connector;
+ }
+
+ private static CdcTargetConnector target(String id) {
+ CdcTargetConnector connector = mock(CdcTargetConnector.class);
+ when(connector.id()).thenReturn(id);
+ return connector;
+ }
+}