Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import org.springframework.stereotype.Component;

import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
Expand All @@ -12,12 +13,21 @@
/**
* Registry of CDC source connector types discovered as Spring beans, plus a safe
* fallback for unit tests without a Spring context.
*
* <p>Connector identifiers are registration authority. Null connectors, blank
* identifiers, and duplicate identifiers are rejected before the registry is
* mutated so one implementation cannot silently replace another.</p>
*/
@Component
public class CdcSourceRegistry {

private final Map<String, CdcSourceConnector> byId = new LinkedHashMap<>();

/**
* Builds a registry from Spring-discovered connector beans.
*
* @param connectors provider of discovered source connectors
*/
public CdcSourceRegistry(ObjectProvider<CdcSourceConnector> connectors) {
connectors.orderedStream().forEach(this::register);
if (byId.isEmpty()) {
Expand All @@ -27,7 +37,10 @@ public CdcSourceRegistry(ObjectProvider<CdcSourceConnector> connectors) {
}

/**
* Explicit list for tests.
* Builds a registry from an explicit connector list, primarily for tests and
* standalone embedding.
*
* @param connectors source connectors to register; a null list is treated as empty
*/
public CdcSourceRegistry(List<CdcSourceConnector> connectors) {
if (connectors != null) {
Expand All @@ -38,19 +51,51 @@ public CdcSourceRegistry(List<CdcSourceConnector> connectors) {
}
}

/**
* Builds a standalone registry with the built-in PostgreSQL/Debezium source.
*/
public CdcSourceRegistry() {
this(List.of());
}

/**
* Registers a CDC source connector without permitting ambiguous authority.
*
* @param connector connector to register
* @throws IllegalArgumentException when the connector is null, its identifier
* is blank, or its identifier is 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 identifier.
*
* @param id connector identifier
* @return the connector when registered, otherwise empty
*/
public Optional<CdcSourceConnector> find(String id) {
return Optional.ofNullable(byId.get(id));
}

/**
* Returns an unmodifiable live view of all registered source connectors in registration order.
*
* @return unmodifiable registered source connectors
*/
public Collection<CdcSourceConnector> all() {
return byId.values();
return Collections.unmodifiableCollection(byId.values());
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -3,32 +3,69 @@
import org.springframework.stereotype.Component;

import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Optional;

/**
* Registry of CDC target types (Kafka, JDBC replica, future warehouses).
*
* <p>Connector identifiers are registration authority. Null connectors, blank
* identifiers, and duplicate identifiers are rejected before the registry is
* mutated so one implementation cannot silently replace another.</p>
*/
@Component
public class CdcTargetRegistry {

private final Map<String, CdcTargetConnector> byId = new LinkedHashMap<>();

/**
* Builds a target registry with the built-in Kafka and JDBC replica targets.
*/
public CdcTargetRegistry() {
register(new KafkaCdcTargetConnector());
register(new JdbcReplicaCdcTargetConnector());
}

/**
* Registers a CDC target connector without permitting ambiguous authority.
*
* @param connector connector to register
* @throws IllegalArgumentException when the connector is null, its identifier
* is blank, or its identifier is 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 identifier.
*
* @param id connector identifier
* @return the connector when registered, otherwise empty
*/
public Optional<CdcTargetConnector> find(String id) {
return Optional.ofNullable(byId.get(id));
}

/**
* Returns an unmodifiable live view of all registered target connectors in registration order.
*
* @return unmodifiable registered target connectors
*/
public Collection<CdcTargetConnector> all() {
return byId.values();
return Collections.unmodifiableCollection(byId.values());
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,118 @@
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.junit.jupiter.api.Assertions.assertTrue;
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());
}

@Test
void sourceConnectorCollectionCannotDeleteRegistrationAuthority() {
CdcSourceRegistry registry = new CdcSourceRegistry(List.of(source("immutable-source")));

assertThrows(UnsupportedOperationException.class, () -> registry.all().clear());
assertTrue(registry.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());
}

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