fix(cdc): preserve DDL log confidentiality replacement

cdc-service/src/main/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplier.java

@@ -24,14 +24,14 @@
  *
  * <p>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.</p>
  */
 @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<String> IDEMPOTENT_DDL_SQL_STATES = Set.of(
             "42P06",
             "42P07",
@@ -46,6 +46,16 @@ public class SchemaChangeReplicaApplier {
     private final Set<String> ddlAllowedPrefixes;
     private final Set<String> 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.
+     *
+     * <p>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.</p>
+     *
+     * @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
@@ -94,20 +116,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,20 +138,21 @@ 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;
     }
 
     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;
     }
 
+    @Nullable
     private String extractDdl(String valueJson) {
         try {
             JsonNode root = objectMapper.readTree(valueJson);
@@ -142,7 +162,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 +250,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 +287,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,

cdc-service/src/test/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplierLoggingTest.java

@@ -0,0 +1,222 @@
+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 java.util.Map;
+
+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 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) " + (char) 0 + secretLiteral;
+        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);
+        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);
+        }
+    }
+}