fix(cdc): preserve valid value when Debezium key JSON is malformed

.dockerignore

@@ -8,3 +8,8 @@
 **/*.swp
 **/*.swo
 **/.DS_Store
+
+# Local environment credentials
+.env
+.env.*
+!.env.example

.gitignore

@@ -38,6 +38,11 @@ package-lock.json
 *.bak
 .~*
 
+# Local environment credentials
+.env
+.env.*
+!.env.example
+
 # OpenCode (local CLI artifacts)
 registered_agents.json
 task_agent_mapping.json

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- Production container builds now use digest-pinned Docker base images while retaining readable Maven/Temurin tags, preventing upstream tag movement from silently changing reviewed build inputs.
 - Durable `POST /api/etl/jobs` submissions now return RFC 9110 `202 Accepted`, a stable pending-job representation, `Location` status-monitor metadata, and explicit replay metadata without changing the synchronous `/api/etl/process` contract. The incomplete intake controller is fail-closed and requires explicit `xtrmetl.etl.jobs.intake-enabled=true` operator opt-in until worker execution and terminal payload clearing are implemented.
 - Concurrent requests using the same authenticated-principal-scoped semantic idempotency key now return immediate RFC 9457 `409 etl_idempotency_request_in_progress` responses through PostgreSQL `pg_try_advisory_xact_lock`; retries after completion still replay the committed response.
 - `POST /api/etl/process` now supports optional authenticated-principal-scoped `Idempotency-Key` retries with atomic target writes, durable response replay, payload-conflict rejection, and explicit replay response metadata.

Dockerfile

@@ -1,4 +1,4 @@
-FROM maven:3.9.13-eclipse-temurin-25 AS build
+FROM maven:3.9.13-eclipse-temurin-25@sha256:ade3c87e3cdfbe04932afa16b31814cbf60b0122d21d78a76530684a1eeb7cc2 AS build
 
 WORKDIR /workspace
 
@@ -18,10 +18,10 @@ RUN mvn -B -DskipTests -pl "${SERVICE}" -am package \
   && mkdir -p /out \
   && cp "${SERVICE}/target/${SERVICE}-"*.jar /out/app.jar
 
-FROM eclipse-temurin:25-jre
+FROM eclipse-temurin:25-jre@sha256:681c543d6f36c50f45e9b5226930a46203dcfa351d3670e9d0bdf0dabae53539
 
 WORKDIR /app
 COPY --from=build --chown=65532:65532 /out/app.jar /app/app.jar
 
 USER 65532:65532
-ENTRYPOINT ["java", "-jar", "/app/app.jar"]
+ENTRYPOINT ["java", "-jar", "/app/app.jar"]

cdc-service/src/main/java/com/xtrmetl/cdc/config/KafkaConfig.java

@@ -23,13 +23,32 @@ public class KafkaConfig {
 
     private static final Logger log = LoggerFactory.getLogger(KafkaConfig.class);
     private static final int MAX_CONCURRENCY = 32;
+    private static final String RETRY_BACKOFF_KEY = "xtrmetl.replica.kafka.retry-backoff-ms";
+    private static final String RETRY_MAX_ATTEMPTS_KEY = "xtrmetl.replica.kafka.retry-max-attempts";
 
+    /**
+     * Builds the replica-listener error handler with bounded retry configuration and terminal
+     * dead-letter recovery.
+     *
+     * <p>Retry settings are deployment-owned. Negative values are rejected before they reach
+     * Spring's {@link FixedBackOff}, while zero remains a valid explicit choice for immediate
+     * retry or no retry attempts.</p>
+     *
+     * @param kafkaTemplate template used to publish exhausted records to the dead-letter topic
+     * @param retryBackoffMs fixed delay between retry attempts in milliseconds; must be non-negative
+     * @param retryMaxAttempts maximum retry attempts after the original delivery; must be non-negative
+     * @return configured listener error handler
+     * @throws IllegalArgumentException when either retry setting is negative
+     */
     @Bean
     public DefaultErrorHandler kafkaListenerErrorHandler(
             @NonNull KafkaTemplate<String, String> kafkaTemplate,
             @Value("${xtrmetl.replica.kafka.retry-backoff-ms:1000}") long retryBackoffMs,
             @Value("${xtrmetl.replica.kafka.retry-max-attempts:30}") long retryMaxAttempts
     ) {
+        requireNonNegative(RETRY_BACKOFF_KEY, retryBackoffMs);
+        requireNonNegative(RETRY_MAX_ATTEMPTS_KEY, retryMaxAttempts);
+
         DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(
                 kafkaTemplate,
                 (record, ex) -> new TopicPartition(record.topic() + ".DLT", record.partition())
@@ -77,4 +96,10 @@ public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerCont
         factory.setCommonErrorHandler(kafkaListenerErrorHandler);
         return factory;
     }
+
+    private static void requireNonNegative(String key, long value) {
+        if (value < 0) {
+            throw new IllegalArgumentException(key + " must be greater than or equal to 0");
+        }
+    }
 }

cdc-service/src/main/java/com/xtrmetl/cdc/config/XtrmetlProperties.java

@@ -105,7 +105,7 @@ public static class Replica {
          */
         private String tables = "processed_data";
         private boolean ddlEnabled = false;
-        private String ddlValidationMode = "none";
+        private String ddlValidationMode = "whitelist";
         private String ddlAllowedPrefixes = "CREATE TABLE,ALTER TABLE,CREATE INDEX";
         private String ddlBlockedPrefixes = "DROP TABLE,DROP SCHEMA,DROP DATABASE,TRUNCATE";
         private final Kafka kafka = new Kafka();

cdc-service/src/main/java/com/xtrmetl/cdc/spi/CanonicalChangeRecord.java

@@ -1,15 +1,20 @@
 package com.xtrmetl.cdc.spi;
 
 import java.util.Collections;
+import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Objects;
 
 /**
- * Product-neutral change event for any-to-any routing (source → targets).
+ * Immutable product-neutral change event for any-to-any CDC routing.
  *
- * <p>Debezium envelopes are adapted via {@link DebeziumChangeRecordMapper}.
- * Not yet wired into the live Kafka publish path — canonical form is for
- * upcoming multi-target routing.</p>
+ * <p>The constructor takes shallow snapshots of the supplied row maps so later
+ * caller mutations cannot change this record's contents, equality, or hash
+ * identity. Null map references become empty maps, while null database values
+ * inside a supplied map are preserved. Debezium envelopes are adapted via
+ * {@link DebeziumChangeRecordMapper}. The canonical form is not yet wired into
+ * the live Kafka publication path and remains the planned multi-target routing
+ * value object.</p>
  */
 public final class CanonicalChangeRecord {
 
@@ -22,6 +27,18 @@ public final class CanonicalChangeRecord {
     private final Map<String, Object> after;
     private final Map<String, Object> pk;
 
+    /**
+     * Creates an immutable canonical CDC record from scalar metadata and row snapshots.
+     *
+     * @param sourceId stable source connector identifier
+     * @param op Debezium-style operation code
+     * @param schema source database schema, when available
+     * @param table source table name, when available
+     * @param tsEpochMs source event timestamp in epoch milliseconds
+     * @param before row values before the change, or {@code null} when unavailable
+     * @param after row values after the change, or {@code null} when unavailable
+     * @param pk primary-key values, or {@code null} when unavailable
+     */
     public CanonicalChangeRecord(
             String sourceId,
             String op,
@@ -37,39 +54,86 @@ public CanonicalChangeRecord(
         this.schema = schema;
         this.table = table;
         this.tsEpochMs = tsEpochMs;
-        this.before = before == null ? Map.of() : Collections.unmodifiableMap(before);
-        this.after = after == null ? Map.of() : Collections.unmodifiableMap(after);
-        this.pk = pk == null ? Map.of() : Collections.unmodifiableMap(pk);
+        this.before = snapshot(before);
+        this.after = snapshot(after);
+        this.pk = snapshot(pk);
     }
 
+    private static Map<String, Object> snapshot(Map<String, Object> source) {
+        if (source == null || source.isEmpty()) {
+            return Map.of();
+        }
+        return Collections.unmodifiableMap(new LinkedHashMap<>(source));
+    }
+
+    /**
+     * Returns the stable source connector identifier.
+     *
+     * @return source connector identifier
+     */
     public String getSourceId() {
         return sourceId;
     }
 
+    /**
+     * Returns the source operation code.
+     *
+     * @return operation code
+     */
     public String getOp() {
         return op;
     }
 
+    /**
+     * Returns the source database schema, when available.
+     *
+     * @return source schema
+     */
     public String getSchema() {
         return schema;
     }
 
+    /**
+     * Returns the source table, when available.
+     *
+     * @return source table
+     */
     public String getTable() {
         return table;
     }
 
+    /**
+     * Returns the source event timestamp in epoch milliseconds.
+     *
+     * @return source event timestamp
+     */
     public long getTsEpochMs() {
         return tsEpochMs;
     }
 
+    /**
+     * Returns the immutable construction-time snapshot of values before the change.
+     *
+     * @return immutable before-values map
+     */
     public Map<String, Object> getBefore() {
         return before;
     }
 
+    /**
+     * Returns the immutable construction-time snapshot of values after the change.
+     *
+     * @return immutable after-values map
+     */
     public Map<String, Object> getAfter() {
         return after;
     }
 
+    /**
+     * Returns the immutable construction-time snapshot of primary-key values.
+     *
+     * @return immutable primary-key map
+     */
     public Map<String, Object> getPk() {
         return pk;
     }

cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcSourceFactory.java

@@ -3,10 +3,12 @@
 import org.springframework.stereotype.Component;
 
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 
 /**
  * Resolves configured CDC source type ids to {@link CdcSourceConnector} instances.
@@ -32,15 +34,23 @@ public Optional<CdcSourceConnector> resolve(String type) {
     }
 
     /**
-     * Validate a multi-source config list. Does not start engines — live capture
-     * remains single-source in {@code CdcService}.
+     * Validates and describes a multi-source configuration list without starting engines.
+     * Live capture remains single-source in {@code CdcService}.
+     *
+     * @param specs configured source entries; {@code null} is treated as an empty list
+     * @return one descriptive row for each configured source
+     * @throws IllegalArgumentException when two entries declare the same source id
      */
     public List<Map<String, Object>> describeConfigured(List<SourceSpec> specs) {
         List<Map<String, Object>> out = new ArrayList<>();
         if (specs == null) {
             return out;
         }
+        Set<String> sourceIds = new HashSet<>();
         for (SourceSpec spec : specs) {
+            if (!sourceIds.add(spec.id())) {
+                throw new IllegalArgumentException("duplicate source id: " + spec.id());
+            }
             Map<String, Object> row = new java.util.LinkedHashMap<>();
             row.put("id", spec.id());
             row.put("type", spec.type());

cdc-service/src/main/java/com/xtrmetl/cdc/spi/DebeziumChangeRecordMapper.java

@@ -27,10 +27,17 @@ public DebeziumChangeRecordMapper(ObjectMapper objectMapper) {
     }
 
     /**
+     * Maps one Debezium value envelope and its optional key into a canonical change record.
+     *
+     * <p>A missing or malformed required value is rejected. A malformed optional key is treated
+     * as unavailable key metadata so a valid value can still supply the existing {@code id}
+     * fallback from its {@code after} or {@code before} object.</p>
+     *
      * @param sourceId logical source id (e.g. {@code postgres-debezium})
-     * @param topic    Kafka / Debezium destination topic ({@code prefix.schema.table})
-     * @param keyJson  optional Debezium key JSON
+     * @param topic Kafka / Debezium destination topic ({@code prefix.schema.table})
+     * @param keyJson optional Debezium key JSON
      * @param valueJson Debezium value JSON
+     * @return the mapped record when the required value envelope is valid, otherwise empty
      */
     public Optional<CanonicalChangeRecord> map(
             String sourceId,
@@ -105,13 +112,17 @@ public Optional<CanonicalChangeRecord> map(
         }
     }
 
-    private Map<String, Object> extractPk(String keyJson) throws IOException {
+    private Map<String, Object> extractPk(String keyJson) {
         if (keyJson == null || keyJson.isBlank()) {
             return Map.of();
         }
-        JsonNode root = objectMapper.readTree(keyJson);
-        JsonNode payload = root.has("payload") ? root.get("payload") : root;
-        return toMap(payload);
+        try {
+            JsonNode root = objectMapper.readTree(keyJson);
+            JsonNode payload = root.has("payload") ? root.get("payload") : root;
+            return toMap(payload);
+        } catch (IOException e) {
+            return Map.of();
+        }
     }
 
     private static String[] schemaTableFromTopic(String topic) {

cdc-service/src/test/java/com/xtrmetl/cdc/config/KafkaConfigTest.java

@@ -18,6 +18,7 @@
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 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.Mockito.mock;
 
@@ -53,6 +54,32 @@ void configuresErrorHandlerWithDeadLetterRecovererAndBackOff() {
         assertFalse(classifier.classify(new IllegalStateException("test")));
     }
 
+    @Test
+    void rejectsNegativeRetryBackoffBeforeBuildingErrorHandler() {
+        KafkaConfig config = new KafkaConfig();
+        KafkaTemplate<String, String> kafkaTemplate = mock(KafkaTemplate.class);
+
+        IllegalArgumentException exception = assertThrows(
+                IllegalArgumentException.class,
+                () -> config.kafkaListenerErrorHandler(kafkaTemplate, -1L, 30L)
+        );
+
+        assertTrue(exception.getMessage().contains("xtrmetl.replica.kafka.retry-backoff-ms"));
+    }
+
+    @Test
+    void rejectsNegativeRetryAttemptsBeforeBuildingErrorHandler() {
+        KafkaConfig config = new KafkaConfig();
+        KafkaTemplate<String, String> kafkaTemplate = mock(KafkaTemplate.class);
+
+        IllegalArgumentException exception = assertThrows(
+                IllegalArgumentException.class,
+                () -> config.kafkaListenerErrorHandler(kafkaTemplate, 1000L, -1L)
+        );
+
+        assertTrue(exception.getMessage().contains("xtrmetl.replica.kafka.retry-max-attempts"));
+    }
+
     @Test
     void configuresListenerFactoryWithRecordAckModeAndCommonErrorHandler() {
         KafkaConfig config = new KafkaConfig();

cdc-service/src/test/java/com/xtrmetl/cdc/config/XtrmetlPropertiesSecurityDefaultTest.java

@@ -0,0 +1,24 @@
+package com.xtrmetl.cdc.config;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+/**
+ * Guards the secure standalone default exposed by the CDC replica configuration object.
+ */
+class XtrmetlPropertiesSecurityDefaultTest {
+
+    @Test
+    void replicaDdlRemainsDisabledAndUsesWhitelistValidationByDefault() {
+        XtrmetlProperties properties = new XtrmetlProperties();
+
+        assertFalse(properties.getReplica().isDdlEnabled(), "DDL replication must remain disabled by default");
+        assertEquals(
+                "whitelist",
+                properties.getReplica().getDdlValidationMode(),
+                "the Java configuration object must match the deployable and metadata secure default"
+        );
+    }
+}