fix(cdc): fail closed on duplicate connector registry IDs

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.
+ *
+ * <p>Connector IDs are unique authority selectors. Duplicate or blank identities fail closed
+ * rather than letting Spring discovery order silently replace an earlier implementation.</p>
  */
 @Component
 public class CdcSourceRegistry {
 
     private final Map<String, CdcSourceConnector> 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<CdcSourceConnector> connectors) {
         connectors.orderedStream().forEach(this::register);
         if (byId.isEmpty()) {
@@ -27,7 +36,10 @@ public CdcSourceRegistry(ObjectProvider<CdcSourceConnector> 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<CdcSourceConnector> connectors) {
         if (connectors != null) {
@@ -38,18 +50,47 @@ public CdcSourceRegistry(List<CdcSourceConnector> 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<CdcSourceConnector> find(String id) {
         return Optional.ofNullable(byId.get(id));
     }
 
+    /**
+     * Returns all registered source connectors in deterministic registration order.
+     *
+     * @return registered source connectors
+     */
     public Collection<CdcSourceConnector> all() {
         return byId.values();
     }

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).
+ *
+ * <p>Connector IDs are unique authority selectors. Duplicate or blank identities fail closed
+ * rather than silently replacing the implementation selected by operator configuration.</p>
  */
 @Component
 public class CdcTargetRegistry {
 
     private final Map<String, CdcTargetConnector> 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<CdcTargetConnector> find(String id) {
         return Optional.ofNullable(byId.get(id));
     }
 
+    /**
+     * Returns all registered target connectors in deterministic registration order.
+     *
+     * @return registered target connectors
+     */
     public Collection<CdcTargetConnector> all() {
         return byId.values();
     }

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;
+    }
+}