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 33d9de67..07466023 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 @@ -8,6 +8,32 @@ */ public interface CdcTargetConnector extends AutoCloseable { + /** + * Describes how the currently shipped product path delivers events to a target. + */ + enum DeliveryMode { + /** Raw Debezium envelopes are published by {@code CdcService} to Kafka. */ + RAW_DEBEZIUM_KAFKA, + /** Processed-data rows are applied by the dedicated JDBC replica pipeline. */ + PROCESSED_DATA_JDBC_REPLICA, + /** Canonical records are delivered directly through this SPI's {@link #write(List)} method. */ + CANONICAL_RECORD_SPI + } + + /** + * Truthful execution metadata for a target connector. + * + * @param productPathLive whether mightyETL currently has a live product path for the target + * @param canonicalWriteSupported whether {@link #write(List)} is wired for canonical records + * @param deliveryMode the execution boundary used by the live product path + */ + record Capabilities( + boolean productPathLive, + boolean canonicalWriteSupported, + DeliveryMode deliveryMode + ) { + } + String id(); String displayName(); @@ -17,6 +43,13 @@ public interface CdcTargetConnector extends AutoCloseable { */ boolean scaffoldOnly(); + /** + * Returns execution metadata without conflating a live legacy/product path with SPI write support. + * + * @return immutable target capability metadata + */ + Capabilities capabilities(); + void validate(Map config); /** 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 6f854cd9..26654369 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 @@ -25,6 +25,16 @@ public boolean scaffoldOnly() { return false; } + /** + * Reports that replica apply is live through the processed-data applier while canonical SPI writes remain unwired. + * + * @return immutable JDBC replica target execution metadata + */ + @Override + public Capabilities capabilities() { + return new Capabilities(true, false, DeliveryMode.PROCESSED_DATA_JDBC_REPLICA); + } + @Override public void validate(Map config) { if (config == null) { 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 62bb059b..5214e310 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 @@ -26,6 +26,16 @@ public boolean scaffoldOnly() { return false; } + /** + * Reports that Kafka is live through the raw Debezium path while canonical SPI writes remain unwired. + * + * @return immutable Kafka target execution metadata + */ + @Override + public Capabilities capabilities() { + return new Capabilities(true, false, DeliveryMode.RAW_DEBEZIUM_KAFKA); + } + @Override public void validate(Map config) { if (config == null) { diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcTargetCapabilityTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcTargetCapabilityTest.java new file mode 100644 index 00000000..32093fa0 --- /dev/null +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcTargetCapabilityTest.java @@ -0,0 +1,43 @@ +package com.xtrmetl.cdc.spi; + +import org.junit.jupiter.api.Test; + +import java.lang.reflect.Method; +import java.util.Arrays; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class CdcTargetCapabilityTest { + + @Test + void liveTargetsDiscloseTheirActualExecutionBoundary() throws Exception { + Method capabilitiesMethod = Arrays.stream(CdcTargetConnector.class.getMethods()) + .filter(method -> method.getName().equals("capabilities")) + .findFirst() + .orElse(null); + + assertTrue(capabilitiesMethod != null, + "CDC target SPI must expose capabilities instead of overloading scaffoldOnly"); + + assertCapabilities( + capabilitiesMethod.invoke(new KafkaCdcTargetConnector()), + "RAW_DEBEZIUM_KAFKA" + ); + assertCapabilities( + capabilitiesMethod.invoke(new JdbcReplicaCdcTargetConnector()), + "PROCESSED_DATA_JDBC_REPLICA" + ); + } + + private static void assertCapabilities(Object capabilities, String expectedDeliveryMode) throws Exception { + Method productPathLive = capabilities.getClass().getMethod("productPathLive"); + Method canonicalWriteSupported = capabilities.getClass().getMethod("canonicalWriteSupported"); + Method deliveryMode = capabilities.getClass().getMethod("deliveryMode"); + + assertTrue((Boolean) productPathLive.invoke(capabilities)); + assertFalse((Boolean) canonicalWriteSupported.invoke(capabilities)); + assertEquals(expectedDeliveryMode, ((Enum) deliveryMode.invoke(capabilities)).name()); + } +}