fix(cdc): acknowledge Kafka delivery before offset progress

ARCHITECTURE.md

@@ -109,9 +109,13 @@ Source DB → CDC Service → Kafka → Consumer Services
 
 1. Database change occurs (INSERT/UPDATE/DELETE)
 2. Debezium captures change from WAL
-3. CDC Service publishes event to Kafka topic
-4. Consumer services process events at their own pace
-5. Loose coupling between producers and consumers
+3. CDC Service submits the raw change event to Kafka
+4. CDC Service waits for the Kafka send future to complete successfully
+5. Only after acknowledgement, Debezium `RecordCommitter` marks that source record processed
+6. The Debezium batch is marked finished only after every record in the batch is processed
+7. Consumer services process events at their own pace
+
+The source-to-Kafka boundary is intentionally fail-closed for source progress: two terminal application send failures leave the current Debezium record unprocessed and fail the batch. Kafka producer idempotence and `acks=all` strengthen producer retry semantics, but they do not turn PostgreSQL, the file-backed Debezium offset store, Kafka, and downstream consumers into one exactly-once transaction. A crash after Kafka acknowledgement but before durable source-offset flush can replay an event; downstream consumers must remain replay-tolerant. See `docs/doctoring/cdc-kafka-acknowledged-delivery.md`.
 
 ## 3. Data Flow Diagrams
 
@@ -212,6 +216,8 @@ Source DB → CDC Service → Kafka → Consumer Services
 └─────────────────────────┘
 ```
 
+The diagram shows the transport topology; the source-progress control loop is stricter than a fire-and-forget arrow. `CdcService` awaits Kafka completion before `RecordCommitter.markProcessed(...)`, and calls `markBatchFinished()` only after the complete batch succeeds. A repeated terminal send failure stops source progress for the affected record. The operator status surface exposes `kafkaPublishSuccess` and `kafkaPublishFailure` attempt counters without payload or credential material.
+
 ### 3.3 Authentication Flow
 
 ```text
@@ -532,6 +538,12 @@ Client Request
 └──────────────────────────────────────────────────────────┘
 ```
 
+#### 8.1.1 Acknowledgement and source-offset boundary
+
+The embedded engine is wired to Debezium's batch `ChangeConsumer` rather than using the one-record callback as the live progress path. For a destination-bearing record, `CdcService` sends to Kafka, waits for the `CompletableFuture<SendResult<...>>`, and only then invokes `RecordCommitter.markProcessed(...)`. The batch is finished only after all records are processed. Two terminal application attempts are permitted; repeated failure aborts the batch without marking the current record. Producer-side `acks=all`, explicit idempotence, and `max.in.flight.requests.per.connection=5` complement this ordering boundary.
+
+This is an at-least-once/replay-tolerant design, not a distributed exactly-once transaction. Debezium source offsets are stored separately from Kafka acknowledgement, so a crash window can replay an already acknowledged event. The full failure model and operator controls are documented in `docs/doctoring/cdc-kafka-acknowledged-delivery.md` and `docs/cdc/ops-and-reliability.md`.
+
 ### 8.2 Spring Retry Mechanism
 
 ```text
@@ -659,6 +671,6 @@ Tuning knobs (replica application, DDL handling, and CDC schema changes):
 
 ---
 
-**Document Version**: 1.0  
-**Last Updated**: 2026-01-08  
+**Document Version**: 1.1  
+**Last Updated**: 2026-08-08  
 **Author**: Technical Architecture Team

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- PostgreSQL→Kafka CDC now waits for Kafka send acknowledgement before advancing Debezium record progress, caps every returned send-future acknowledgement wait at 65 seconds per application attempt, retries one terminal publication failure or timeout without marking the source record, fails the batch after a repeated failure, and requires `acks=all`, producer idempotence, bounded delivery/block timeouts, and at most five in-flight requests per connection. The boundary remains explicitly replay-tolerant rather than an end-to-end exactly-once claim.
 - 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.
@@ -22,6 +23,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- CDC source-publication reliability evidence and operator diagnostics: `kafkaPublishSuccess` / `kafkaPublishFailure` status counters, deterministic acknowledgement/failure/interrupt/timeout tests, finite 65-second returned-future waits, architecture and operations guidance, and APA 7 primary references in `docs/doctoring/cdc-kafka-acknowledged-delivery.md`.
 - Principal-scoped durable asynchronous ETL job intake and owner-scoped status resources, Flyway `etl_job_records` migration, deterministic replay/conflict coverage, and the explicit worker boundary in `docs/etl/durable-job-intake.md`.
 - Durable idempotency ledger migration, PostgreSQL transaction advisory-lock adapter, deterministic concurrency/rollback coverage, and the operator/client contract `docs/etl/idempotent-retries.md`.
 - ETL problem-details client and operator contract: `docs/api/problem-details.md`.
@@ -249,5 +251,5 @@ This changelog will be updated:
 ---
 
 **Changelog Version**: 1.0  
-**Last Updated**: 2026-08-04  
+**Last Updated**: 2026-08-08  
 **Maintained By**: Development Team

cdc-service/src/main/java/com/xtrmetl/cdc/service/CdcService.java

@@ -8,31 +8,39 @@
 import io.debezium.engine.ChangeEvent;
 import io.debezium.engine.DebeziumEngine;
 import io.debezium.engine.format.Json;
+import org.apache.kafka.common.KafkaException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.context.event.ApplicationReadyEvent;
 import org.springframework.context.event.EventListener;
 import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.SendResult;
 import org.springframework.lang.Nullable;
 import org.springframework.stereotype.Service;
 
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicLong;
 
 @Service
 public class CdcService implements DisposableBean {
 
     private static final Logger log = LoggerFactory.getLogger(CdcService.class);
+    private static final int KAFKA_PUBLISH_ATTEMPTS = 2;
+    private static final long KAFKA_ACKNOWLEDGEMENT_WAIT_MS = 65_000L;
 
     private ExecutorService executor;
     private final KafkaTemplate<String, String> kafkaTemplate;
@@ -42,10 +50,25 @@ public class CdcService implements DisposableBean {
     private final DebeziumChangeRecordMapper changeRecordMapper;
     private final AtomicLong canonicalMapSuccess = new AtomicLong();
     private final AtomicLong canonicalMapFailure = new AtomicLong();
+    private final AtomicLong kafkaPublishSuccess = new AtomicLong();
+    private final AtomicLong kafkaPublishFailure = new AtomicLong();
 
     private DebeziumEngine<ChangeEvent<String, String>> debeziumEngine;
     private Future<?> engineTask;
 
+    /**
+     * Constructs the CDC service with explicit startup, canonical-mapping, and change-mapping dependencies.
+     *
+     * <p>The Kafka template is the publication boundary used by both the compatibility send path and the
+     * acknowledgement-aware Debezium batch path. Automatic startup controls whether the engine starts after
+     * Spring reports application readiness. Canonical mapping is optional observation logic and never grants
+     * permission to advance a Debezium offset.</p>
+     *
+     * @param kafkaTemplate Kafka producer facade used to publish raw Debezium change events
+     * @param autoStart whether the Debezium engine should start after application readiness
+     * @param canonicalMapEnabled whether optional canonical-record mapping should run before publication
+     * @param changeRecordMapper optional canonical mapper; may be {@code null} when mapping is disabled
+     */
     public CdcService(
             KafkaTemplate<String, String> kafkaTemplate,
             @Value("${xtrmetl.cdc.autostart:true}") boolean autoStart,
@@ -59,7 +82,10 @@ public CdcService(
     }
 
     /**
-     * Test helper / backward-compatible constructor.
+     * Constructs a CDC service with canonical mapping disabled for compatibility callers and focused tests.
+     *
+     * @param kafkaTemplate Kafka producer facade used to publish raw Debezium change events
+     * @param autoStart whether the Debezium engine should start after application readiness
      */
     public CdcService(KafkaTemplate<String, String> kafkaTemplate, boolean autoStart) {
         this(kafkaTemplate, autoStart, false, null);
@@ -78,6 +104,12 @@ public void maybeAutoStart() {
         }
     }
 
+    /**
+     * Handles Spring Boot application readiness by applying the configured CDC auto-start policy.
+     *
+     * <p>This lifecycle callback delegates to {@link #maybeAutoStart()}; it does not bypass the configured
+     * auto-start flag and is safe when startup is intentionally disabled for operator-controlled deployments.</p>
+     */
     @EventListener(ApplicationReadyEvent.class)
     public void onApplicationReady() {
         maybeAutoStart();
@@ -152,6 +184,11 @@ public synchronized boolean isRunning() {
         return engineTask != null && !engineTask.isDone();
     }
 
+    /**
+     * Returns whether automatic CDC startup is enabled.
+     *
+     * @return {@code true} when the application-readiness callback should start the Debezium engine
+     */
     public boolean isAutoStart() {
         return autoStart;
     }
@@ -177,6 +214,8 @@ public synchronized Map<String, Object> getStatus() {
         status.put("canonicalMapEnabled", canonicalMapEnabled);
         status.put("canonicalMapSuccess", canonicalMapSuccess.get());
         status.put("canonicalMapFailure", canonicalMapFailure.get());
+        status.put("kafkaPublishSuccess", kafkaPublishSuccess.get());
+        status.put("kafkaPublishFailure", kafkaPublishFailure.get());
         status.put("configPrefixes", "mightyetl.* (preferred) or xtrmetl.* (legacy); dual-read via EnvironmentPostProcessor");
         status.put("notes", "Capture is PostgreSQL→Kafka only; see docs/cdc/any-to-any-cdc.md");
         return status;
@@ -221,43 +260,120 @@ public synchronized void shutdown() {
     private void initializeDebeziumEngine() {
         this.debeziumEngine = DebeziumEngine.create(Json.class)
                 .using(getCdcConfiguration().asProperties())
-                .notifying(this::handleChangeEvent)
+                .notifying(this::handleChangeBatch)
                 .build();
     }
 
+    /**
+     * Releases CDC engine and executor resources when the Spring bean is destroyed.
+     *
+     * <p>This framework lifecycle hook delegates to {@link #shutdown()} so explicit shutdown and container-driven
+     * destruction share the same bounded termination and interrupt handling.</p>
+     */
     @Override
     public void destroy() {
         shutdown();
     }
 
     /**
-     * Publishes Debezium key/value JSON to the destination Kafka topic.
-     *
-     * Debezium에서 받은 CDC 변경 이벤트의 JSON key/value를 해당 topic으로 전송한다.
+     * Publishes one Debezium change event without waiting for acknowledgement.
      *
-     * Uses a key-less send overload when the key is absent, following Spring Kafka conventions.
+     * <p>This method is retained for direct callers and compatibility tests. The live Debezium engine uses
+     * {@link #handleChangeBatch(List, DebeziumEngine.RecordCommitter)} so source offsets are advanced only after
+     * Kafka acknowledges publication.</p>
      *
-     * key가 없으면 Spring Kafka 계약에 맞게 key-less send 오버로드를 사용한다.
-     *
-     * @param changeEvent Debezium의 변경 이벤트로부터 key/value와 destination을 포함하는 이벤트 객체
+     * @param changeEvent Debezium change event containing destination, key, and JSON value
      */
     protected void handleChangeEvent(ChangeEvent<String, String> changeEvent) {
         String topic = changeEvent.destination();
         if (topic == null) {
             return;
         }
 
-        String key = changeEvent.key();
-        String value = changeEvent.value();
+        maybeMapCanonical(topic, changeEvent.key(), changeEvent.value());
+        sendChangeEvent(changeEvent);
+    }
+
+    /**
+     * Publishes a Debezium batch to Kafka and advances source offsets only after broker acknowledgement.
+     *
+     * <p>Each event with a destination is sent as raw Debezium JSON, awaited for at most 65 seconds per
+     * application attempt, and retried once after a terminal publication failure or acknowledgement timeout.
+     * A record is marked processed only after an acknowledged send. The batch is marked finished only after every
+     * record has been processed. An interrupt propagates immediately without advancing the current record or batch.</p>
+     *
+     * <p>An event without a destination is treated as non-publishable engine metadata and is marked processed
+     * without touching Kafka or the publication counters.</p>
+     *
+     * @param records ordered Debezium change events for the current engine batch
+     * @param committer Debezium offset committer controlling record and batch progress
+     * @throws InterruptedException when the worker is interrupted while awaiting Kafka acknowledgement
+     * @throws KafkaException when Kafka does not acknowledge an event within two application-level attempts
+     */
+    protected void handleChangeBatch(
+            List<ChangeEvent<String, String>> records,
+            DebeziumEngine.RecordCommitter<ChangeEvent<String, String>> committer
+    ) throws InterruptedException {
+        for (ChangeEvent<String, String> changeEvent : records) {
+            String topic = changeEvent.destination();
+            if (topic != null) {
+                maybeMapCanonical(topic, changeEvent.key(), changeEvent.value());
+                publishWithAcknowledgement(changeEvent);
+            }
+            committer.markProcessed(changeEvent);
+        }
+        committer.markBatchFinished();
+    }
+
+    /**
+     * Sends an event and waits for Kafka acknowledgement, retrying once after a terminal failure.
+     */
+    private void publishWithAcknowledgement(ChangeEvent<String, String> changeEvent) throws InterruptedException {
+        try {
+            awaitAcknowledgement(changeEvent);
+            kafkaPublishSuccess.incrementAndGet();
+            return;
+        } catch (ExecutionException firstFailure) {
+            kafkaPublishFailure.incrementAndGet();
+        }
+
+        try {
+            awaitAcknowledgement(changeEvent);
+            kafkaPublishSuccess.incrementAndGet();
+        } catch (ExecutionException secondFailure) {
+            kafkaPublishFailure.incrementAndGet();
+            throw new KafkaException(
+                    "Kafka publication failed after " + KAFKA_PUBLISH_ATTEMPTS + " attempts",
+                    secondFailure.getCause()
+            );
+        }
+    }
 
-        maybeMapCanonical(topic, key, value);
+    /**
+     * Performs one Kafka send attempt and waits interruptibly for at most 65 seconds for acknowledgement.
+     */
+    private void awaitAcknowledgement(ChangeEvent<String, String> changeEvent)
+            throws InterruptedException, ExecutionException {
+        try {
+            sendChangeEvent(changeEvent).get(KAFKA_ACKNOWLEDGEMENT_WAIT_MS, TimeUnit.MILLISECONDS);
+        } catch (TimeoutException timeoutException) {
+            throw new ExecutionException(timeoutException);
+        } catch (RuntimeException runtimeException) {
+            throw new ExecutionException(runtimeException);
+        }
+    }
 
-        // Live path: raw Debezium JSON (not canonical) for consumer compatibility.
+    /**
+     * Selects the keyed or keyless Spring Kafka send overload for one raw Debezium JSON event.
+     */
+    private CompletableFuture<SendResult<String, String>> sendChangeEvent(ChangeEvent<String, String> changeEvent) {
+        String topic = changeEvent.destination();
+        String key = changeEvent.key();
+        String value = changeEvent.value();
         if (key != null) {
-            kafkaTemplate.send(topic, key, value);
-        } else {
-            kafkaTemplate.send(topic, value);
+            return kafkaTemplate.send(topic, key, value);
         }
+        return kafkaTemplate.send(topic, value);
     }
 
     /**

cdc-service/src/main/resources/application.yml

@@ -13,6 +13,12 @@ spring:
     producer:
       key-serializer: org.apache.kafka.common.serialization.StringSerializer
       value-serializer: org.apache.kafka.common.serialization.StringSerializer
+      acks: all
+      properties:
+        "[enable.idempotence]": true
+        "[delivery.timeout.ms]": ${CDC_KAFKA_DELIVERY_TIMEOUT_MS:60000}
+        "[max.block.ms]": ${CDC_KAFKA_MAX_BLOCK_MS:30000}
+        "[max.in.flight.requests.per.connection]": 5
     consumer:
       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer