From 9c724e3f6f888827157af470aa100fbe709ceb26 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 14:16:22 +0900 Subject: [PATCH 1/5] test(cdc): prove replicated DDL stays out of logs --- ...SchemaChangeReplicaApplierLoggingTest.java | 169 ++++++++++++++++++ 1 file changed, 169 insertions(+) create mode 100644 cdc-service/src/test/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplierLoggingTest.java diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplierLoggingTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplierLoggingTest.java new file mode 100644 index 00000000..f68f2468 --- /dev/null +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplierLoggingTest.java @@ -0,0 +1,169 @@ +package com.xtrmetl.cdc.replication; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.springframework.boot.test.system.CapturedOutput; +import org.springframework.boot.test.system.OutputCaptureExtension; +import org.springframework.dao.DataAccessException; +import org.springframework.jdbc.core.JdbcTemplate; + +import java.sql.SQLException; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +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; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; + +/** + * Verifies that schema-replication observability never republishes raw DDL or driver diagnostics. + */ +@ExtendWith(OutputCaptureExtension.class) +class SchemaChangeReplicaApplierLoggingTest { + + @Test + void successfulApplyLogsOnlyBoundedMetadataNotRawDdl(CapturedOutput output) { + JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class); + SchemaChangeReplicaApplier applier = applier(jdbcTemplate, "none", "", ""); + String secretLiteral = "buyer-contract-secret-8472"; + String inputDdl = "CREATE TABLE confidential_record(secret_value text DEFAULT '" + secretLiteral + "')"; + String executedDdl = "CREATE TABLE IF NOT EXISTS confidential_record(secret_value text DEFAULT '" + + secretLiteral + "')"; + + applier.apply(schemaTopic(), null, ddlEnvelope(inputDdl)); + + verify(jdbcTemplate).execute(eq(executedDdl)); + assertSafeLogs(output, "Applied schema change DDL on replica", secretLiteral, "DEFAULT", "ddl="); + } + + @Test + void blockedDdlLogsPolicyOutcomeWithoutRawStatement(CapturedOutput output) { + JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class); + SchemaChangeReplicaApplier applier = applier( + jdbcTemplate, + "whitelist", + "CREATE TABLE,ALTER TABLE,CREATE INDEX", + "" + ); + String secretPath = "/srv/private/buyer-contract-secret-8472"; + String ddl = "CREATE TABLESPACE reporting LOCATION '" + secretPath + "'"; + + assertThrows(IllegalArgumentException.class, () -> applier.apply(schemaTopic(), null, ddlEnvelope(ddl))); + + verifyNoInteractions(jdbcTemplate); + assertSafeLogs(output, "Blocked DDL by validation policy", secretPath, "TABLESPACE", "ddl="); + } + + @Test + void duplicateDdlLogsOutcomeWithoutSqlOrDriverDiagnostics(CapturedOutput output) { + JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class); + SchemaChangeReplicaApplier applier = applier(jdbcTemplate, "none", "", ""); + String secretLiteral = "buyer-contract-secret-8472"; + String inputDdl = "CREATE TABLE confidential_record(secret_value text DEFAULT '" + secretLiteral + "')"; + String executedDdl = "CREATE TABLE IF NOT EXISTS confidential_record(secret_value text DEFAULT '" + + secretLiteral + "')"; + SQLException sqlException = new SQLException( + "relation already exists at jdbc:postgresql://db.internal/prod?password=driver-secret", + "42P07" + ); + DataAccessException duplicate = new DataAccessException("driver-secret", sqlException) {}; + doThrow(duplicate).when(jdbcTemplate).execute(eq(executedDdl)); + + assertDoesNotThrow(() -> applier.apply(schemaTopic(), null, ddlEnvelope(inputDdl))); + + verify(jdbcTemplate).execute(eq(executedDdl)); + assertSafeLogs( + output, + "Schema change DDL already applied; skipping duplicate", + secretLiteral, + "driver-secret", + "jdbc:postgresql://", + "ddl=" + ); + } + + @Test + void executionFailureLogsOutcomeWithoutSqlOrDriverDiagnostics(CapturedOutput output) { + JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class); + SchemaChangeReplicaApplier applier = applier(jdbcTemplate, "none", "", ""); + String secretLiteral = "buyer-contract-secret-8472"; + String inputDdl = "CREATE TABLE confidential_record(secret_value text DEFAULT '" + secretLiteral + "')"; + String executedDdl = "CREATE TABLE IF NOT EXISTS confidential_record(secret_value text DEFAULT '" + + secretLiteral + "')"; + DataAccessException failure = new DataAccessException( + "jdbc:postgresql://db.internal/prod?password=driver-secret" + ) {}; + doThrow(failure).when(jdbcTemplate).execute(eq(executedDdl)); + + DataAccessException thrown = assertThrows( + DataAccessException.class, + () -> applier.apply(schemaTopic(), null, ddlEnvelope(inputDdl)) + ); + + assertSame(failure, thrown); + assertSafeLogs( + output, + "Failed to apply schema change DDL on replica", + secretLiteral, + "driver-secret", + "jdbc:postgresql://", + "ddl=" + ); + } + + @Test + void malformedEventLogsOnlyStableParseClassification(CapturedOutput output) { + JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class); + SchemaChangeReplicaApplier applier = applier(jdbcTemplate, "none", "", ""); + String malformed = "{buyer-contract-secret-8472"; + + assertDoesNotThrow(() -> applier.apply(schemaTopic(), null, malformed)); + + verifyNoInteractions(jdbcTemplate); + assertSafeLogs( + output, + "Failed to parse Debezium schema change JSON; skipping DDL apply", + "buyer-contract-secret-8472", + "JsonParseException", + "ReaderBasedJsonParser" + ); + } + + private static SchemaChangeReplicaApplier applier( + JdbcTemplate jdbcTemplate, + String validationMode, + String allowedPrefixes, + String blockedPrefixes + ) { + return new SchemaChangeReplicaApplier( + jdbcTemplate, + new ObjectMapper(), + true, + validationMode, + allowedPrefixes, + blockedPrefixes + ); + } + + private static String schemaTopic() { + return "xtrmetl-cdc.schema-changes"; + } + + private static String ddlEnvelope(String ddl) { + return "{\"payload\":{\"ddl\":\"" + ddl.replace("'", "\\u0027") + "\"}}"; + } + + private static void assertSafeLogs(CapturedOutput output, String expected, String... forbidden) { + String logs = output.getOut() + output.getErr(); + assertTrue(logs.contains(expected)); + for (String value : forbidden) { + assertFalse(logs.contains(value), () -> "Log output exposed forbidden value: " + value); + } + } +} From 017d1355dc04876a0f03e0d655b7fd3a4d3a9d9b Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 14:39:09 +0900 Subject: [PATCH 2/5] fix(cdc): keep DDL diagnostics out of logs --- .../SchemaChangeReplicaApplier.java | 39 ++++++------------- 1 file changed, 12 insertions(+), 27 deletions(-) diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplier.java b/cdc-service/src/main/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplier.java index c9f016f8..50228a0a 100644 --- a/cdc-service/src/main/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplier.java +++ b/cdc-service/src/main/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplier.java @@ -24,14 +24,14 @@ * *

DDL execution is disabled by default. When enabled, the default policy is a positive * allow-list, only a single comment-free statement is accepted, and prefix matches must end - * at a SQL token boundary. + * at a SQL token boundary. Ordinary lifecycle logs retain bounded outcome metadata without + * reproducing raw DDL statements or provider exception diagnostics.

*/ @Service @ConditionalOnProperty(prefix = "xtrmetl.replica", name = "enabled", havingValue = "true") public class SchemaChangeReplicaApplier { private static final Logger log = LoggerFactory.getLogger(SchemaChangeReplicaApplier.class); - private static final int DDL_LOG_MAX_LENGTH = 500; private static final Set IDEMPOTENT_DDL_SQL_STATES = Set.of( "42P06", "42P07", @@ -94,20 +94,17 @@ public void apply(@Nullable String topic, @Nullable String keyJson, @Nullable St // single-statement, comment-free, and configured policy gates above. jdbcTemplate.execute(ddl); // nosemgrep: java.spring.security.audit.spring-sqli.spring-sqli if (log.isInfoEnabled()) { - log.info("Applied schema change DDL on replica (topic={}, ddl={})", - topic, truncateForLog(ddl)); + log.info("Applied schema change DDL on replica (topic={})", topic); } } catch (DataAccessException e) { if (isIdempotentDuplicate(e)) { if (log.isInfoEnabled()) { - log.info("Schema change DDL already applied; skipping duplicate (topic={}, ddl={})", - topic, truncateForLog(ddl)); + log.info("Schema change DDL already applied; skipping duplicate (topic={})", topic); } return; } if (log.isErrorEnabled()) { - log.error("Failed to apply schema change DDL on replica (topic={}, ddl={})", - topic, truncateForLog(ddl), e); + log.error("Failed to apply schema change DDL on replica (topic={})", topic); } throw e; } @@ -119,7 +116,7 @@ private String requireSingleStatement(String topic, String ddl) { trimmed = trimmed.substring(0, trimmed.length() - 1).trim(); } if (trimmed.contains(";")) { - logBlocked(topic, ddl, "Blocked multi-statement DDL"); + logBlocked(topic, "Blocked multi-statement DDL"); throw new IllegalArgumentException("Multiple SQL statements are not allowed"); } return trimmed; @@ -127,7 +124,7 @@ private String requireSingleStatement(String topic, String ddl) { private String requireCommentFree(String topic, String ddl) { if (ddl.contains("--") || ddl.contains("/*") || ddl.contains("*/") || ddl.indexOf('\0') >= 0) { - logBlocked(topic, ddl, "Blocked DDL containing SQL comments or NUL"); + logBlocked(topic, "Blocked DDL containing SQL comments or NUL"); throw new IllegalArgumentException("SQL comments and NUL characters are not allowed in replicated DDL"); } return ddl; @@ -142,7 +139,7 @@ private String extractDdl(String valueJson) { } return payload.path("ddl").asText(null); } catch (IOException e) { - log.warn("Failed to parse Debezium schema change JSON; skipping DDL apply", e); + log.warn("Failed to parse Debezium schema change JSON; skipping DDL apply"); return null; } } @@ -230,21 +227,20 @@ private void validateDdl(String topic, String ddl) { String normalized = normalizeForValidation(ddl); if (ddlValidationMode == DdlValidationMode.BLOCKLIST && ddlBlockedPrefixes.stream().anyMatch(prefix -> matchesPrefix(normalized, prefix))) { - logBlocked(topic, ddl, "Blocked DDL by validation policy"); + logBlocked(topic, "Blocked DDL by validation policy"); throw new IllegalArgumentException("DDL blocked by validation policy"); } if (ddlValidationMode == DdlValidationMode.WHITELIST && ddlAllowedPrefixes.stream().noneMatch(prefix -> matchesPrefix(normalized, prefix))) { - logBlocked(topic, ddl, "Blocked DDL by validation policy"); + logBlocked(topic, "Blocked DDL by validation policy"); throw new IllegalArgumentException("DDL blocked by validation policy"); } } - private void logBlocked(String topic, String ddl, String message) { + private void logBlocked(String topic, String message) { if (log.isWarnEnabled()) { - log.warn("{} (mode={}, topic={}, ddl={})", - message, ddlValidationMode, topic, truncateForLog(ddl)); + log.warn("{} (mode={}, topic={})", message, ddlValidationMode, topic); } } @@ -268,17 +264,6 @@ private static String normalizeForValidation(String ddl) { return ddl.trim().replaceAll("\\s+", " ").toUpperCase(Locale.ROOT); } - private static String truncateForLog(String ddl) { - if (ddl == null) { - return null; - } - String normalized = ddl.trim().replaceAll("\\s+", " "); - if (normalized.length() <= DDL_LOG_MAX_LENGTH) { - return normalized; - } - return normalized.substring(0, DDL_LOG_MAX_LENGTH) + "..."; - } - private enum DdlValidationMode { NONE, WHITELIST, From 2d6a651a07a30750186e0421feb6696c0216eb3f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 15:40:27 +0900 Subject: [PATCH 3/5] test(cdc): cover blocked DDL log confidentiality --- ...SchemaChangeReplicaApplierLoggingTest.java | 53 +++++++++++++++++++ 1 file changed, 53 insertions(+) diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplierLoggingTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplierLoggingTest.java index f68f2468..db50896e 100644 --- a/cdc-service/src/test/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplierLoggingTest.java +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplierLoggingTest.java @@ -9,6 +9,7 @@ import org.springframework.jdbc.core.JdbcTemplate; import java.sql.SQLException; +import java.util.Map; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertFalse; @@ -60,6 +61,58 @@ void blockedDdlLogsPolicyOutcomeWithoutRawStatement(CapturedOutput output) { assertSafeLogs(output, "Blocked DDL by validation policy", secretPath, "TABLESPACE", "ddl="); } + @Test + void multiStatementBlockLogsClassificationWithoutRawStatement(CapturedOutput output) { + JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class); + SchemaChangeReplicaApplier applier = applier(jdbcTemplate, "none", "", ""); + String secretLiteral = "buyer-contract-secret-8472"; + String ddl = "CREATE TABLE confidential_record(id int); DROP TABLE " + secretLiteral; + + assertThrows(IllegalArgumentException.class, () -> applier.apply(schemaTopic(), null, ddlEnvelope(ddl))); + + verifyNoInteractions(jdbcTemplate); + assertSafeLogs(output, "Blocked multi-statement DDL", secretLiteral, "DROP TABLE", "ddl="); + } + + @Test + void sqlCommentBlockLogsClassificationWithoutRawStatement(CapturedOutput output) { + JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class); + SchemaChangeReplicaApplier applier = applier(jdbcTemplate, "none", "", ""); + String secretLiteral = "buyer-contract-secret-8472"; + String ddl = "CREATE TABLE confidential_record(id int) -- " + secretLiteral; + + assertThrows(IllegalArgumentException.class, () -> applier.apply(schemaTopic(), null, ddlEnvelope(ddl))); + + verifyNoInteractions(jdbcTemplate); + assertSafeLogs( + output, + "Blocked DDL containing SQL comments or NUL", + secretLiteral, + "confidential_record", + "ddl=" + ); + } + + @Test + void nulBlockLogsClassificationWithoutRawStatement(CapturedOutput output) throws Exception { + JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class); + SchemaChangeReplicaApplier applier = applier(jdbcTemplate, "none", "", ""); + String secretLiteral = "buyer-contract-secret-8472"; + String ddl = "CREATE TABLE confidential_record(id int) " + secretLiteral + (char) 0; + String envelope = new ObjectMapper().writeValueAsString(Map.of("payload", Map.of("ddl", ddl))); + + assertThrows(IllegalArgumentException.class, () -> applier.apply(schemaTopic(), null, envelope)); + + verifyNoInteractions(jdbcTemplate); + assertSafeLogs( + output, + "Blocked DDL containing SQL comments or NUL", + secretLiteral, + "confidential_record", + "ddl=" + ); + } + @Test void duplicateDdlLogsOutcomeWithoutSqlOrDriverDiagnostics(CapturedOutput output) { JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class); From 45e140969977299a5cfe7cac2fe6b0e8689b034f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 15:49:06 +0900 Subject: [PATCH 4/5] test(cdc): exercise retained NUL log guard --- .../cdc/replication/SchemaChangeReplicaApplierLoggingTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplierLoggingTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplierLoggingTest.java index db50896e..0c65f67b 100644 --- a/cdc-service/src/test/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplierLoggingTest.java +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplierLoggingTest.java @@ -98,7 +98,7 @@ void nulBlockLogsClassificationWithoutRawStatement(CapturedOutput output) throws JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class); SchemaChangeReplicaApplier applier = applier(jdbcTemplate, "none", "", ""); String secretLiteral = "buyer-contract-secret-8472"; - String ddl = "CREATE TABLE confidential_record(id int) " + secretLiteral + (char) 0; + String ddl = "CREATE TABLE confidential_record(id int) " + (char) 0 + secretLiteral; String envelope = new ObjectMapper().writeValueAsString(Map.of("payload", Map.of("ddl", ddl))); assertThrows(IllegalArgumentException.class, () -> applier.apply(schemaTopic(), null, envelope)); From 1b8e93e92f8f6f1943d13e4bcb899d46bf77641b Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 19:12:54 +0900 Subject: [PATCH 5/5] docs(cdc): preserve schema applier API contracts --- .../SchemaChangeReplicaApplier.java | 23 +++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplier.java b/cdc-service/src/main/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplier.java index 50228a0a..4df7007f 100644 --- a/cdc-service/src/main/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplier.java +++ b/cdc-service/src/main/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplier.java @@ -46,6 +46,16 @@ public class SchemaChangeReplicaApplier { private final Set ddlAllowedPrefixes; private final Set ddlBlockedPrefixes; + /** + * Creates the schema-change applier from replica JDBC and DDL-policy configuration. + * + * @param jdbcTemplate replica-database JDBC access + * @param objectMapper JSON parser used for Debezium event envelopes + * @param ddlEnabled whether schema-changing SQL may be executed at all + * @param ddlValidationMode configured validation mode: {@code whitelist}, {@code blocklist}, or {@code none} + * @param ddlAllowedPrefixes comma-separated positive prefixes used by whitelist mode + * @param ddlBlockedPrefixes comma-separated prohibited prefixes used by blocklist mode + */ public SchemaChangeReplicaApplier( @Qualifier("replicaJdbcTemplate") JdbcTemplate jdbcTemplate, ObjectMapper objectMapper, @@ -70,6 +80,18 @@ public SchemaChangeReplicaApplier( } } + /** + * Applies one schema-change event when DDL replication is enabled and the event passes policy. + * + *

Events outside the schema-change topic family, blank payloads, and envelopes without DDL + * are ignored. Malformed JSON is classified as an unavailable event and skipped without + * publishing parser diagnostics. Policy violations fail closed with an exception before JDBC + * execution. Non-duplicate JDBC failures are rethrown after a bounded diagnostic event.

+ * + * @param topic source Kafka topic; only the schema-change suffix is accepted + * @param keyJson optional Debezium key, currently unused by schema application + * @param valueJson Debezium value envelope containing a {@code ddl} field + */ public void apply(@Nullable String topic, @Nullable String keyJson, @Nullable String valueJson) { if (!ddlEnabled || topic == null @@ -130,6 +152,7 @@ private String requireCommentFree(String topic, String ddl) { return ddl; } + @Nullable private String extractDdl(String valueJson) { try { JsonNode root = objectMapper.readTree(valueJson);