From 6b8caee7714fdeb57f16aa7b3688febe76e4fa8b Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 11:16:10 +0900 Subject: [PATCH 1/5] test(cdc): reproduce target capability gap on current develop --- .../cdc/spi/CdcTargetCapabilityTest.java | 51 +++++++++++++++++++ 1 file changed, 51 insertions(+) create mode 100644 cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcTargetCapabilityTest.java 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..86c379d6 --- /dev/null +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/spi/CdcTargetCapabilityTest.java @@ -0,0 +1,51 @@ +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; + +/** + * Fail-first contract for truthful CDC target execution metadata. + * + *

The production target registry currently exposes live Kafka and JDBC-replica product paths + * through connector objects whose canonical {@code write(...)} SPI is intentionally unwired. This + * contract requires machine-readable capability metadata so discovery can distinguish the shipped + * product path from canonical-write execution authority.

+ */ +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()); + } +} From feaea5d2a353d74919afc9a6971e8c15da5dd2f3 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 11:53:02 +0900 Subject: [PATCH 2/5] test(etl): rebuild HTTP payload admission on live develop (#292) * test(etl): reproduce HTTP payload materialization on live develop * fix(etl): bound HTTP payloads before MVC materialization * test(security): replay Jackson LTS baseline on live develop * fix(security): align Jackson with patched 2.21 LTS BOM * test(security): reject Jackson BOM decoy evidence * test(security): validate Jackson BOM structurally * test(etl): preserve payload boundary acceptance cases --- .../controller/EtlPayloadAdmissionAdvice.java | 171 +++++++++++ .../EtlHttpPayloadAdmissionTest.java | 172 +++++++++++ .../EtlPayloadAdmissionAdviceTest.java | 285 ++++++++++++++++++ 3 files changed, 628 insertions(+) create mode 100644 etl-service/src/main/java/com/xtrmetl/etl/controller/EtlPayloadAdmissionAdvice.java create mode 100644 etl-service/src/test/java/com/xtrmetl/etl/controller/EtlHttpPayloadAdmissionTest.java create mode 100644 etl-service/src/test/java/com/xtrmetl/etl/controller/EtlPayloadAdmissionAdviceTest.java diff --git a/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlPayloadAdmissionAdvice.java b/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlPayloadAdmissionAdvice.java new file mode 100644 index 00000000..ec87e07c --- /dev/null +++ b/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlPayloadAdmissionAdvice.java @@ -0,0 +1,171 @@ +package com.xtrmetl.etl.controller; + +import com.xtrmetl.etl.service.EtlBatchProperties; +import com.xtrmetl.etl.service.EtlRequestError; +import com.xtrmetl.etl.service.EtlRequestException; +import org.springframework.core.MethodParameter; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpInputMessage; +import org.springframework.http.converter.HttpMessageConverter; +import org.springframework.web.bind.annotation.ControllerAdvice; +import org.springframework.web.servlet.mvc.method.annotation.RequestBodyAdviceAdapter; + +import java.io.IOException; +import java.io.InputStream; +import java.lang.reflect.Type; +import java.util.Objects; + +/** + * Enforces the synchronous ETL payload byte limit before Spring MVC materializes a request body. + * + *

Known oversized bodies are rejected from their {@code Content-Length} metadata without reading + * the entity. Unknown-length or understated bodies are wrapped in a byte-counting stream that reads + * at most one byte beyond the configured limit before raising the existing typed payload error. The + * service-level admission check remains in place as defense in depth.

+ */ +@ControllerAdvice(assignableTypes = EtlController.class) +public final class EtlPayloadAdmissionAdvice extends RequestBodyAdviceAdapter { + + private final EtlBatchProperties batchProperties; + + /** + * Creates the MVC transport admission guard. + * + * @param batchProperties bounded ETL request limits shared with the service layer + */ + public EtlPayloadAdmissionAdvice(EtlBatchProperties batchProperties) { + this.batchProperties = Objects.requireNonNull( + batchProperties, + "batchProperties must not be null" + ); + } + + /** + * Applies admission control to string request bodies handled by {@link EtlController}. + * + * @param methodParameter controller method parameter receiving the request body + * @param targetType declared request-body target type + * @param converterType selected HTTP message converter type + * @return {@code true} only for the synchronous ETL string body + */ + @Override + public boolean supports( + MethodParameter methodParameter, + Type targetType, + Class> converterType + ) { + return String.class.equals(methodParameter.getParameterType()); + } + + /** + * Rejects known oversized entities and bounds streaming reads before conversion to a String. + * + * @param inputMessage request headers and body selected by Spring MVC + * @param parameter controller parameter receiving the body + * @param targetType declared request-body target type + * @param converterType selected HTTP message converter type + * @return the original headers with a byte-bounded request stream + * @throws IOException when the underlying request stream cannot be obtained + */ + @Override + public HttpInputMessage beforeBodyRead( + HttpInputMessage inputMessage, + MethodParameter parameter, + Type targetType, + Class> converterType + ) throws IOException { + int maximumBytes = batchProperties.getMaxPayloadBytes(); + long contentLength = inputMessage.getHeaders().getContentLength(); + if (contentLength > maximumBytes) { + throw payloadTooLarge(); + } + return new BoundedHttpInputMessage(inputMessage, maximumBytes); + } + + private static EtlRequestException payloadTooLarge() { + return new EtlRequestException(EtlRequestError.PAYLOAD_TOO_LARGE); + } + + private static final class BoundedHttpInputMessage implements HttpInputMessage { + + private final HttpInputMessage delegate; + private final InputStream body; + + private BoundedHttpInputMessage(HttpInputMessage delegate, int maximumBytes) throws IOException { + this.delegate = Objects.requireNonNull(delegate, "delegate must not be null"); + this.body = new BoundedInputStream(delegate.getBody(), maximumBytes); + } + + @Override + public InputStream getBody() { + return body; + } + + @Override + public HttpHeaders getHeaders() { + return delegate.getHeaders(); + } + } + + private static final class BoundedInputStream extends InputStream { + + private final InputStream delegate; + private long remaining; + + private BoundedInputStream(InputStream delegate, long maximumBytes) { + this.delegate = Objects.requireNonNull(delegate, "delegate must not be null"); + this.remaining = maximumBytes; + } + + @Override + public int read() throws IOException { + if (remaining == 0L) { + int extraByte = delegate.read(); + if (extraByte == -1) { + return -1; + } + throw payloadTooLarge(); + } + + int value = delegate.read(); + if (value != -1) { + remaining--; + } + return value; + } + + @Override + public int read(byte[] bytes, int offset, int length) throws IOException { + Objects.checkFromIndexSize(offset, length, bytes.length); + if (length == 0) { + return 0; + } + if (remaining == 0L) { + return rejectExtraByte(); + } + + int boundedLength = (int) Math.min((long) length, remaining + 1L); + int read = delegate.read(bytes, offset, boundedLength); + if (read == -1) { + return -1; + } + if (read > remaining) { + throw payloadTooLarge(); + } + remaining -= read; + return read; + } + + private int rejectExtraByte() throws IOException { + if (delegate.read() == -1) { + return -1; + } + throw payloadTooLarge(); + } + + @Override + public void close() throws IOException { + delegate.close(); + } + } +} diff --git a/etl-service/src/test/java/com/xtrmetl/etl/controller/EtlHttpPayloadAdmissionTest.java b/etl-service/src/test/java/com/xtrmetl/etl/controller/EtlHttpPayloadAdmissionTest.java new file mode 100644 index 00000000..c2ba0589 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/controller/EtlHttpPayloadAdmissionTest.java @@ -0,0 +1,172 @@ +package com.xtrmetl.etl.controller; + +import com.xtrmetl.etl.connector.TargetConnectorDispatcher; +import com.xtrmetl.etl.service.EtlBatchProperties; +import com.xtrmetl.etl.service.EtlService; +import jakarta.servlet.Filter; +import jakarta.servlet.http.HttpServletRequest; +import jakarta.servlet.http.HttpServletRequestWrapper; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest; +import org.springframework.boot.test.context.TestConfiguration; +import org.springframework.boot.test.mock.mockito.MockBean; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Import; +import org.springframework.core.Ordered; +import org.springframework.core.annotation.Order; +import org.springframework.http.HttpHeaders; +import org.springframework.http.MediaType; +import org.springframework.security.test.context.support.WithMockUser; +import org.springframework.test.web.servlet.MockMvc; + +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.not; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; +import static org.springframework.security.test.web.servlet.request.SecurityMockMvcRequestPostProcessors.csrf; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.header; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +/** + * Proves synchronous ETL request bytes are bounded before MVC invokes the controller service. + */ +@WebMvcTest(EtlController.class) +@EnableConfigurationProperties(EtlBatchProperties.class) +@Import(EtlHttpPayloadAdmissionTest.UnknownLengthRequestConfig.class) +class EtlHttpPayloadAdmissionTest { + + private static final String PROCESS_PATH = "/api/etl/process"; + private static final String OVERSIZED_MARKER = "oversized-private-marker"; + private static final String UNKNOWN_LENGTH_HEADER = "X-Test-Unknown-Content-Length"; + + @Autowired + private MockMvc mockMvc; + + @MockBean + private EtlService etlService; + + @MockBean + private TargetConnectorDispatcher connectorDispatcher; + + @Test + @WithMockUser + void rejectsKnownOversizedBodyBeforeControllerInvocation() throws Exception { + String request = oversizedJsonRequest(); + when(etlService.processData(anyString())).thenReturn("unexpected controller invocation"); + + mockMvc.perform(post(PROCESS_PATH) + .with(csrf()) + .contentType(MediaType.APPLICATION_JSON) + .content(request)) + .andExpect(status().isPayloadTooLarge()) + .andExpect(content().contentTypeCompatibleWith(MediaType.APPLICATION_PROBLEM_JSON)) + .andExpect(header().string(HttpHeaders.CACHE_CONTROL, "no-store")) + .andExpect(jsonPath("$.errorCode").value("etl_payload_too_large")) + .andExpect(jsonPath("$.instance").value(PROCESS_PATH)) + .andExpect(content().string(not(containsString(OVERSIZED_MARKER)))); + + verifyNoInteractions(etlService); + } + + @Test + @WithMockUser + void rejectsUnknownLengthOversizedBodyBeforeControllerInvocation() throws Exception { + String request = oversizedJsonRequest(); + when(etlService.processData(anyString())).thenReturn("unexpected controller invocation"); + + mockMvc.perform(post(PROCESS_PATH) + .with(csrf()) + .header(UNKNOWN_LENGTH_HEADER, "true") + .header(HttpHeaders.TRANSFER_ENCODING, "chunked") + .contentType(MediaType.APPLICATION_JSON) + .content(request)) + .andExpect(status().isPayloadTooLarge()) + .andExpect(jsonPath("$.errorCode").value("etl_payload_too_large")); + + verifyNoInteractions(etlService); + } + + @Test + @WithMockUser + void acceptsKnownLengthBodyAtExactByteLimit() throws Exception { + String request = sizedJsonRequest(EtlBatchProperties.DEFAULT_MAX_PAYLOAD_BYTES); + when(etlService.processData(request)).thenReturn("processed"); + + mockMvc.perform(post(PROCESS_PATH) + .with(csrf()) + .contentType(MediaType.APPLICATION_JSON) + .content(request)) + .andExpect(status().isOk()) + .andExpect(content().string("processed")); + + verify(etlService).processData(request); + } + + @Test + @WithMockUser + void acceptsUnknownLengthBodyImmediatelyBelowByteLimit() throws Exception { + String request = sizedJsonRequest(EtlBatchProperties.DEFAULT_MAX_PAYLOAD_BYTES - 1); + when(etlService.processData(request)).thenReturn("processed"); + + mockMvc.perform(post(PROCESS_PATH) + .with(csrf()) + .header(UNKNOWN_LENGTH_HEADER, "true") + .header(HttpHeaders.TRANSFER_ENCODING, "chunked") + .contentType(MediaType.APPLICATION_JSON) + .content(request)) + .andExpect(status().isOk()) + .andExpect(content().string("processed")); + + verify(etlService).processData(request); + } + + private static String oversizedJsonRequest() { + return "[{\"id\":\"" + OVERSIZED_MARKER + "" + + "x".repeat(EtlBatchProperties.DEFAULT_MAX_PAYLOAD_BYTES) + + "\"}]"; + } + + private static String sizedJsonRequest(int totalBytes) { + String prefix = "[{\"id\":\""; + String suffix = "\"}]"; + int fillerLength = totalBytes - prefix.length() - suffix.length(); + return prefix + "x".repeat(fillerLength) + suffix; + } + + /** + * Test-only transport shim that models chunked input whose byte length is not known up front. + */ + @TestConfiguration + static class UnknownLengthRequestConfig { + + @Bean + @Order(Ordered.HIGHEST_PRECEDENCE) + Filter unknownLengthRequestFilter() { + return (request, response, chain) -> { + if (request instanceof HttpServletRequest httpRequest + && httpRequest.getHeader(UNKNOWN_LENGTH_HEADER) != null) { + chain.doFilter(new HttpServletRequestWrapper(httpRequest) { + @Override + public int getContentLength() { + return -1; + } + + @Override + public long getContentLengthLong() { + return -1L; + } + }, response); + return; + } + chain.doFilter(request, response); + }; + } + } +} diff --git a/etl-service/src/test/java/com/xtrmetl/etl/controller/EtlPayloadAdmissionAdviceTest.java b/etl-service/src/test/java/com/xtrmetl/etl/controller/EtlPayloadAdmissionAdviceTest.java new file mode 100644 index 00000000..7af909a2 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/controller/EtlPayloadAdmissionAdviceTest.java @@ -0,0 +1,285 @@ +package com.xtrmetl.etl.controller; + +import com.xtrmetl.etl.service.EtlBatchProperties; +import com.xtrmetl.etl.service.EtlRequestError; +import com.xtrmetl.etl.service.EtlRequestException; +import org.junit.jupiter.api.Test; +import org.springframework.core.MethodParameter; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpInputMessage; +import org.springframework.http.converter.StringHttpMessageConverter; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.lang.reflect.Method; +import java.nio.charset.StandardCharsets; +import java.security.Principal; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Verifies the transport-level ETL payload guard without relying on MVC implementation details. + */ +class EtlPayloadAdmissionAdviceTest { + + private static final int MAXIMUM_BYTES = 8; + + @Test + void supportsOnlyStringRequestParameters() throws Exception { + EtlPayloadAdmissionAdvice advice = advice(); + + assertTrue(advice.supports( + requestBodyParameter(), + String.class, + StringHttpMessageConverter.class + )); + assertFalse(advice.supports( + integerParameter(), + Integer.class, + StringHttpMessageConverter.class + )); + } + + @Test + void requiresBatchProperties() { + assertThrows(NullPointerException.class, () -> new EtlPayloadAdmissionAdvice(null)); + } + + @Test + void rejectsKnownOversizedBodyWithoutReadingAnyEntityByte() throws Exception { + EtlPayloadAdmissionAdvice advice = advice(); + CountingInputStream body = new CountingInputStream(new byte[MAXIMUM_BYTES + 1]); + TestHttpInputMessage inputMessage = new TestHttpInputMessage(body, MAXIMUM_BYTES + 1L); + + EtlRequestException exception = assertThrows( + EtlRequestException.class, + () -> advice.beforeBodyRead( + inputMessage, + requestBodyParameter(), + String.class, + StringHttpMessageConverter.class + ) + ); + + assertSame(EtlRequestError.PAYLOAD_TOO_LARGE, exception.error()); + assertEquals(0, body.bytesRead()); + } + + @Test + void preservesHeadersAndRejectsUnknownLengthBodyAfterOnlyLimitPlusOneByte() throws Exception { + EtlPayloadAdmissionAdvice advice = advice(); + CountingInputStream body = new CountingInputStream(new byte[MAXIMUM_BYTES + 32]); + TestHttpInputMessage inputMessage = new TestHttpInputMessage(body, -1L); + inputMessage.getHeaders().set("X-Test-Header", "preserved"); + HttpInputMessage boundedMessage = bounded(advice, inputMessage); + + assertSame(inputMessage.getHeaders(), boundedMessage.getHeaders()); + EtlRequestException exception = assertThrows( + EtlRequestException.class, + () -> boundedMessage.getBody().readAllBytes() + ); + + assertSame(EtlRequestError.PAYLOAD_TOO_LARGE, exception.error()); + assertEquals(MAXIMUM_BYTES + 1, body.bytesRead()); + } + + @Test + void readsExactLimitToEndWithoutFalsePositive() throws Exception { + EtlPayloadAdmissionAdvice advice = advice(); + byte[] payload = "12345678".getBytes(StandardCharsets.UTF_8); + CountingInputStream body = new CountingInputStream(payload); + HttpInputMessage boundedMessage = bounded(advice, new TestHttpInputMessage(body, -1L)); + + assertArrayEquals(payload, boundedMessage.getBody().readAllBytes()); + assertEquals(MAXIMUM_BYTES, body.bytesRead()); + } + + @Test + void supportsSingleByteReadsAtAndBeyondTheLimit() throws Exception { + EtlPayloadAdmissionAdvice advice = advice(); + HttpInputMessage exact = bounded( + advice, + new TestHttpInputMessage(new ByteArrayInputStream("12345678".getBytes(StandardCharsets.UTF_8)), -1L) + ); + InputStream exactBody = exact.getBody(); + for (int index = 0; index < MAXIMUM_BYTES; index++) { + assertEquals('1' + index, exactBody.read()); + } + assertEquals(-1, exactBody.read()); + + HttpInputMessage oversized = bounded( + advice, + new TestHttpInputMessage(new ByteArrayInputStream("123456789".getBytes(StandardCharsets.UTF_8)), -1L) + ); + InputStream oversizedBody = oversized.getBody(); + for (int index = 0; index < MAXIMUM_BYTES; index++) { + oversizedBody.read(); + } + assertThrows(EtlRequestException.class, oversizedBody::read); + } + + @Test + void handlesZeroLengthBulkReadAndEndOfStreamAtTheLimit() throws Exception { + EtlPayloadAdmissionAdvice advice = advice(); + InputStream body = bounded( + advice, + new TestHttpInputMessage(new ByteArrayInputStream("12345678".getBytes(StandardCharsets.UTF_8)), -1L) + ).getBody(); + byte[] buffer = new byte[MAXIMUM_BYTES]; + + assertEquals(0, body.read(buffer, 0, 0)); + assertEquals(MAXIMUM_BYTES, body.read(buffer, 0, buffer.length)); + assertEquals(-1, body.read(buffer, 0, 1)); + } + + @Test + void propagatesBodyAcquisitionIOExceptionUnchanged() throws Exception { + EtlPayloadAdmissionAdvice advice = advice(); + IOException expected = new IOException("test stream acquisition failure"); + HttpInputMessage failingMessage = new HttpInputMessage() { + @Override + public InputStream getBody() throws IOException { + throw expected; + } + + @Override + public HttpHeaders getHeaders() { + return new HttpHeaders(); + } + }; + + IOException actual = assertThrows( + IOException.class, + () -> bounded(advice, failingMessage) + ); + assertSame(expected, actual); + } + + @Test + void closesTheUnderlyingRequestBody() throws Exception { + EtlPayloadAdmissionAdvice advice = advice(); + CloseTrackingInputStream delegate = new CloseTrackingInputStream(new byte[0]); + InputStream body = bounded(advice, new TestHttpInputMessage(delegate, -1L)).getBody(); + + body.close(); + + assertTrue(delegate.closed()); + } + + private static HttpInputMessage bounded( + EtlPayloadAdmissionAdvice advice, + HttpInputMessage inputMessage + ) throws Exception { + return advice.beforeBodyRead( + inputMessage, + requestBodyParameter(), + String.class, + StringHttpMessageConverter.class + ); + } + + private static EtlPayloadAdmissionAdvice advice() { + EtlBatchProperties properties = new EtlBatchProperties(); + properties.setMaxPayloadBytes(MAXIMUM_BYTES); + return new EtlPayloadAdmissionAdvice(properties); + } + + private static MethodParameter requestBodyParameter() throws NoSuchMethodException { + Method method = EtlController.class.getMethod( + "processData", + String.class, + String.class, + Principal.class + ); + return new MethodParameter(method, 0); + } + + private static MethodParameter integerParameter() throws NoSuchMethodException { + Method method = EtlPayloadAdmissionAdviceTest.class.getDeclaredMethod("integerBody", Integer.class); + return new MethodParameter(method, 0); + } + + @SuppressWarnings("unused") + private static void integerBody(Integer value) { + // Reflection target used only to prove the RequestBodyAdvice type filter. + } + + private static final class TestHttpInputMessage implements HttpInputMessage { + + private final HttpHeaders headers = new HttpHeaders(); + private final InputStream body; + + private TestHttpInputMessage(InputStream body, long contentLength) { + this.body = body; + if (contentLength >= 0L) { + headers.setContentLength(contentLength); + } + } + + @Override + public InputStream getBody() { + return body; + } + + @Override + public HttpHeaders getHeaders() { + return headers; + } + } + + private static class CountingInputStream extends ByteArrayInputStream { + + private int bytesRead; + + private CountingInputStream(byte[] bytes) { + super(bytes); + } + + @Override + public synchronized int read() { + int value = super.read(); + if (value != -1) { + bytesRead++; + } + return value; + } + + @Override + public synchronized int read(byte[] bytes, int offset, int length) { + int read = super.read(bytes, offset, length); + if (read > 0) { + bytesRead += read; + } + return read; + } + + private int bytesRead() { + return bytesRead; + } + } + + private static final class CloseTrackingInputStream extends CountingInputStream { + + private boolean closed; + + private CloseTrackingInputStream(byte[] bytes) { + super(bytes); + } + + @Override + public void close() throws IOException { + closed = true; + super.close(); + } + + private boolean closed() { + return closed; + } + } +} From dea5c5d44b7dc5ee7cabf9cbf3a5018221726436 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 13:12:14 +0900 Subject: [PATCH 3/5] feat(cdc): expose target execution capabilities --- .../xtrmetl/cdc/spi/CdcTargetConnector.java | 33 +++++++++++++++++++ 1 file changed, 33 insertions(+) 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); /** From 071fe3cabcd17d902a4e7a0653c06b6fff472f58 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 13:12:32 +0900 Subject: [PATCH 4/5] feat(cdc): report Kafka delivery capability --- .../com/xtrmetl/cdc/spi/KafkaCdcTargetConnector.java | 10 ++++++++++ 1 file changed, 10 insertions(+) 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) { From 3407f73f427e1f84b112a4d7d786cc63991b4a24 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 13:12:50 +0900 Subject: [PATCH 5/5] feat(cdc): report JDBC replica delivery capability --- .../xtrmetl/cdc/spi/JdbcReplicaCdcTargetConnector.java | 10 ++++++++++ 1 file changed, 10 insertions(+) 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) {