From f2683286c4a768ead11da0895a511ab2bec9af88 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 10 Aug 2026 11:11:17 +0900 Subject: [PATCH 1/4] test(cdc): reject ambiguous connector registry identities --- .../cdc/spi/CdcRegistryIdentityTest.java | 76 +++++++++++++++++++ 1 file changed, 76 insertions(+) create mode 100644 cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcRegistryIdentityTest.java 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..171db789 --- /dev/null +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcRegistryIdentityTest.java @@ -0,0 +1,76 @@ +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 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; + } +} From c19e59777f19e8a23dd02ceb8f9543e421733abd Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 10 Aug 2026 11:16:36 +0900 Subject: [PATCH 2/4] fix(cdc): reject ambiguous source connector identities --- .../xtrmetl/cdc/spi/CdcSourceRegistry.java | 45 ++++++++++++++++++- 1 file changed, 43 insertions(+), 2 deletions(-) diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcSourceRegistry.java b/cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcSourceRegistry.java index 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(); } From 28b01de58d0c24cec864ef63bd9b9f3697c936d1 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 10 Aug 2026 11:17:14 +0900 Subject: [PATCH 3/4] fix(cdc): reject ambiguous target connector identities --- .../xtrmetl/cdc/spi/CdcTargetRegistry.java | 34 ++++++++++++++++++- 1 file changed, 33 insertions(+), 1 deletion(-) 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(); } From 1ba5527effcc4605b44d53effd0dfb3be3618164 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 10 Aug 2026 11:32:30 +0900 Subject: [PATCH 4/4] test(cdc): cover null connector identity rejection --- .../cdc/spi/CdcRegistryIdentityTest.java | 24 +++++++++++++++++++ 1 file changed, 24 insertions(+) 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 171db789..8d6ad0c1 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 @@ -37,6 +37,30 @@ void duplicateTargetConnectorIdsFailClosedInsteadOfReplacingRegistration() { 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(" ");