diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 412d0b98..0ad8f3b1 100644 --- a/ARCHITECTURE.md +++ b/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>`, 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 diff --git a/CHANGELOG.md b/CHANGELOG.md index 45b08b8c..b3ff7a7f 100644 --- a/CHANGELOG.md +++ b/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 \ No newline at end of file diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/service/CdcService.java b/cdc-service/src/main/java/com/xtrmetl/cdc/service/CdcService.java index d33b6525..995e8cd4 100644 --- a/cdc-service/src/main/java/com/xtrmetl/cdc/service/CdcService.java +++ b/cdc-service/src/main/java/com/xtrmetl/cdc/service/CdcService.java @@ -8,6 +8,7 @@ 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; @@ -15,6 +16,7 @@ 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; @@ -22,17 +24,23 @@ 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 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> debeziumEngine; private Future engineTask; + /** + * Constructs the CDC service with explicit startup, canonical-mapping, and change-mapping dependencies. + * + *

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.

+ * + * @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 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 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. + * + *

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.

+ */ @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 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,25 +260,29 @@ 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. + * + *

This framework lifecycle hook delegates to {@link #shutdown()} so explicit shutdown and container-driven + * destruction share the same bounded termination and interrupt handling.

+ */ @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. + *

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.

* - * 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 changeEvent) { String topic = changeEvent.destination(); @@ -247,17 +290,90 @@ protected void handleChangeEvent(ChangeEvent changeEvent) { 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. + * + *

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.

+ * + *

An event without a destination is treated as non-publishable engine metadata and is marked processed + * without touching Kafka or the publication counters.

+ * + * @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> records, + DebeziumEngine.RecordCommitter> committer + ) throws InterruptedException { + for (ChangeEvent 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 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 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> sendChangeEvent(ChangeEvent 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); } /** diff --git a/cdc-service/src/main/resources/application.yml b/cdc-service/src/main/resources/application.yml index b299ebc9..c492e7fa 100644 --- a/cdc-service/src/main/resources/application.yml +++ b/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 diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/service/CdcKafkaPublishAcknowledgementTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/service/CdcKafkaPublishAcknowledgementTest.java new file mode 100644 index 00000000..9ebc5d97 --- /dev/null +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/service/CdcKafkaPublishAcknowledgementTest.java @@ -0,0 +1,273 @@ +package com.xtrmetl.cdc.service; + +import io.debezium.engine.ChangeEvent; +import io.debezium.engine.DebeziumEngine; +import org.apache.kafka.common.KafkaException; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.support.SendResult; + +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.List; +import java.util.concurrent.CompletableFuture; +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.AtomicReference; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.timeout; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Guards the PostgreSQL-to-Kafka delivery boundary so Debezium offsets are not marked processed + * before Kafka has acknowledged the corresponding change event. + */ +class CdcKafkaPublishAcknowledgementTest { + + private KafkaTemplate kafkaTemplate; + private CdcService cdcService; + + @BeforeEach + void setUp() { + @SuppressWarnings("unchecked") + KafkaTemplate template = (KafkaTemplate) mock(KafkaTemplate.class); + kafkaTemplate = template; + cdcService = new CdcService(kafkaTemplate, false); + } + + @Test + void waitsForBrokerAcknowledgementBeforeMarkingDebeziumOffset() throws Exception { + ChangeEvent event = event("orders.customer_updates", "customer-42", "{\"op\":\"u\"}"); + CompletableFuture> pendingAcknowledgement = new CompletableFuture<>(); + when(kafkaTemplate.send("orders.customer_updates", "customer-42", "{\"op\":\"u\"}")) + .thenReturn(pendingAcknowledgement); + + DebeziumEngine.RecordCommitter> committer = committer(); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future processing = executor.submit(() -> { + try { + cdcService.handleChangeBatch(List.of(event), committer); + } catch (InterruptedException interruptedException) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(interruptedException); + } + }); + + verify(kafkaTemplate, timeout(1_000)).send( + "orders.customer_updates", "customer-42", "{\"op\":\"u\"}" + ); + assertThrows(TimeoutException.class, () -> processing.get(100, TimeUnit.MILLISECONDS)); + verify(committer, never()).markProcessed(any()); + verify(committer, never()).markBatchFinished(); + + pendingAcknowledgement.complete(mock(SendResult.class)); + processing.get(1, TimeUnit.SECONDS); + + var ordered = inOrder(committer); + ordered.verify(committer).markProcessed(event); + ordered.verify(committer).markBatchFinished(); + assertEquals(1L, cdcService.getStatus().get("kafkaPublishSuccess")); + assertEquals(0L, cdcService.getStatus().get("kafkaPublishFailure")); + } finally { + executor.shutdownNow(); + } + } + + @Test + void retriesFailedKafkaPublicationBeforeOffsetCommit() throws Exception { + ChangeEvent event = event("orders.customer_updates", "customer-42", "{\"op\":\"u\"}"); + CompletableFuture> failed = new CompletableFuture<>(); + failed.completeExceptionally(new KafkaException("broker unavailable")); + CompletableFuture> succeeded = + CompletableFuture.completedFuture(mock(SendResult.class)); + when(kafkaTemplate.send("orders.customer_updates", "customer-42", "{\"op\":\"u\"}")) + .thenReturn(failed, succeeded); + + DebeziumEngine.RecordCommitter> committer = committer(); + cdcService.handleChangeBatch(List.of(event), committer); + + verify(kafkaTemplate, times(2)).send( + "orders.customer_updates", "customer-42", "{\"op\":\"u\"}" + ); + verify(committer).markProcessed(event); + verify(committer).markBatchFinished(); + assertEquals(1L, cdcService.getStatus().get("kafkaPublishSuccess")); + assertEquals(1L, cdcService.getStatus().get("kafkaPublishFailure")); + } + + @Test + void exhaustsBoundedKafkaPublicationAttemptsWithoutOffsetCommit() throws Exception { + ChangeEvent event = event("orders.customer_updates", "customer-42", "{\"op\":\"u\"}"); + CompletableFuture> firstFailure = new CompletableFuture<>(); + firstFailure.completeExceptionally(new KafkaException("broker unavailable")); + CompletableFuture> secondFailure = new CompletableFuture<>(); + secondFailure.completeExceptionally(new KafkaException("broker still unavailable")); + when(kafkaTemplate.send("orders.customer_updates", "customer-42", "{\"op\":\"u\"}")) + .thenReturn(firstFailure, secondFailure); + + DebeziumEngine.RecordCommitter> committer = committer(); + + KafkaException failure = assertThrows( + KafkaException.class, + () -> cdcService.handleChangeBatch(List.of(event), committer) + ); + + assertTrue(failure.getMessage().contains("2 attempts")); + verify(kafkaTemplate, times(2)).send( + "orders.customer_updates", "customer-42", "{\"op\":\"u\"}" + ); + verify(committer, never()).markProcessed(any()); + verify(committer, never()).markBatchFinished(); + assertEquals(0L, cdcService.getStatus().get("kafkaPublishSuccess")); + assertEquals(2L, cdcService.getStatus().get("kafkaPublishFailure")); + } + + @Test + void retriesSynchronousKafkaSendFailureBeforeOffsetCommit() throws Exception { + ChangeEvent event = event("orders.customer_updates", "customer-42", "{\"op\":\"u\"}"); + CompletableFuture> succeeded = + CompletableFuture.completedFuture(mock(SendResult.class)); + when(kafkaTemplate.send("orders.customer_updates", "customer-42", "{\"op\":\"u\"}")) + .thenThrow(new KafkaException("producer temporarily unavailable")) + .thenReturn(succeeded); + + DebeziumEngine.RecordCommitter> committer = committer(); + cdcService.handleChangeBatch(List.of(event), committer); + + verify(kafkaTemplate, times(2)).send( + "orders.customer_updates", "customer-42", "{\"op\":\"u\"}" + ); + verify(committer).markProcessed(event); + verify(committer).markBatchFinished(); + assertEquals(1L, cdcService.getStatus().get("kafkaPublishSuccess")); + assertEquals(1L, cdcService.getStatus().get("kafkaPublishFailure")); + } + + @Test + void keylessEventIsAcknowledgedBeforeOffsetCommit() throws Exception { + ChangeEvent event = event("orders.customer_updates", null, "{\"op\":\"d\"}"); + when(kafkaTemplate.send("orders.customer_updates", "{\"op\":\"d\"}")) + .thenReturn(CompletableFuture.completedFuture(mock(SendResult.class))); + + DebeziumEngine.RecordCommitter> committer = committer(); + cdcService.handleChangeBatch(List.of(event), committer); + + verify(kafkaTemplate).send("orders.customer_updates", "{\"op\":\"d\"}"); + verify(committer).markProcessed(event); + verify(committer).markBatchFinished(); + } + + @Test + void eventWithoutDestinationAdvancesWithoutPublishing() throws Exception { + ChangeEvent event = event(null, "customer-42", "{\"op\":\"u\"}"); + DebeziumEngine.RecordCommitter> committer = committer(); + + cdcService.handleChangeBatch(List.of(event), committer); + + verify(kafkaTemplate, never()).send(any(String.class), any(String.class)); + verify(kafkaTemplate, never()).send(any(String.class), any(String.class), any(String.class)); + verify(committer).markProcessed(event); + verify(committer).markBatchFinished(); + assertEquals(0L, cdcService.getStatus().get("kafkaPublishSuccess")); + assertEquals(0L, cdcService.getStatus().get("kafkaPublishFailure")); + } + + @Test + void interruptedAcknowledgementStopsBatchWithoutMarkingOffset() throws Exception { + ChangeEvent event = event("orders.customer_updates", "customer-42", "{\"op\":\"u\"}"); + CompletableFuture> pendingAcknowledgement = new CompletableFuture<>(); + when(kafkaTemplate.send("orders.customer_updates", "customer-42", "{\"op\":\"u\"}")) + .thenReturn(pendingAcknowledgement); + DebeziumEngine.RecordCommitter> committer = committer(); + AtomicReference failure = new AtomicReference<>(); + + Thread processing = new Thread(() -> { + try { + cdcService.handleChangeBatch(List.of(event), committer); + } catch (Throwable throwable) { + failure.set(throwable); + } + }, "cdc-kafka-ack-test"); + processing.start(); + verify(kafkaTemplate, timeout(1_000)).send( + "orders.customer_updates", "customer-42", "{\"op\":\"u\"}" + ); + + processing.interrupt(); + processing.join(1_000); + + assertFalse(processing.isAlive(), "interrupted CDC publication must stop promptly"); + assertInstanceOf(InterruptedException.class, failure.get()); + verify(committer, never()).markProcessed(any()); + verify(committer, never()).markBatchFinished(); + } + + @Test + void producerConfigurationRequiresDurableAcknowledgedDelivery() throws Exception { + String configuration = Files.readString( + projectRoot().resolve("cdc-service/src/main/resources/application.yml"), + StandardCharsets.UTF_8 + ).replace("\r\n", "\n"); + + assertTrue(configuration.contains("acks: all")); + assertTrue(configuration.contains("\"[enable.idempotence]\": true")); + assertTrue(configuration.contains( + "\"[delivery.timeout.ms]\": ${CDC_KAFKA_DELIVERY_TIMEOUT_MS:60000}" + )); + assertTrue(configuration.contains( + "\"[max.block.ms]\": ${CDC_KAFKA_MAX_BLOCK_MS:30000}" + )); + } + + @SuppressWarnings("unchecked") + private static DebeziumEngine.RecordCommitter> committer() { + return (DebeziumEngine.RecordCommitter>) mock( + DebeziumEngine.RecordCommitter.class + ); + } + + @SuppressWarnings("unchecked") + private static ChangeEvent event(String destination, String key, String value) { + ChangeEvent event = (ChangeEvent) mock(ChangeEvent.class); + when(event.destination()).thenReturn(destination); + when(event.key()).thenReturn(key); + when(event.value()).thenReturn(value); + return event; + } + + private static Path projectRoot() { + Path current = Paths.get(System.getProperty("user.dir")).toAbsolutePath(); + Path lastPom = null; + while (current != null) { + if (Files.exists(current.resolve(".git"))) { + return current; + } + if (Files.exists(current.resolve("pom.xml"))) { + lastPom = current; + } + current = current.getParent(); + } + if (lastPom != null) { + return lastPom; + } + throw new IllegalStateException("project root not found"); + } +} diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/service/CdcKafkaPublishTimeoutTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/service/CdcKafkaPublishTimeoutTest.java new file mode 100644 index 00000000..7fc7ea8a --- /dev/null +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/service/CdcKafkaPublishTimeoutTest.java @@ -0,0 +1,94 @@ +package com.xtrmetl.cdc.service; + +import io.debezium.engine.ChangeEvent; +import io.debezium.engine.DebeziumEngine; +import org.apache.kafka.common.KafkaException; +import org.junit.jupiter.api.Test; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.support.SendResult; + +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Proves that a Kafka send future which never reaches a terminal state cannot stall CDC offset + * handling indefinitely. + */ +class CdcKafkaPublishTimeoutTest { + + private static final long EXPECTED_ACKNOWLEDGEMENT_WAIT_MS = 65_000L; + + /** + * Requires each application-level Kafka attempt to have a finite acknowledgement wait while + * preserving fail-closed Debezium offset semantics. + * + * @throws Exception when mocked future or committer contracts surface checked failures + */ + @Test + void timesOutHungKafkaAcknowledgementsWithoutAdvancingOffsets() throws Exception { + @SuppressWarnings("unchecked") + KafkaTemplate kafkaTemplate = + (KafkaTemplate) mock(KafkaTemplate.class); + @SuppressWarnings("unchecked") + CompletableFuture> firstHungSend = + (CompletableFuture>) mock(CompletableFuture.class); + @SuppressWarnings("unchecked") + CompletableFuture> secondHungSend = + (CompletableFuture>) mock(CompletableFuture.class); + + when(firstHungSend.get(EXPECTED_ACKNOWLEDGEMENT_WAIT_MS, TimeUnit.MILLISECONDS)) + .thenThrow(new TimeoutException("first acknowledgement stalled")); + when(secondHungSend.get(EXPECTED_ACKNOWLEDGEMENT_WAIT_MS, TimeUnit.MILLISECONDS)) + .thenThrow(new TimeoutException("second acknowledgement stalled")); + when(kafkaTemplate.send("orders.customer_updates", "customer-42", "{\"op\":\"u\"}")) + .thenReturn(firstHungSend, secondHungSend); + + CdcService cdcService = new CdcService(kafkaTemplate, false); + ChangeEvent event = event(); + DebeziumEngine.RecordCommitter> committer = committer(); + + KafkaException failure = assertThrows( + KafkaException.class, + () -> cdcService.handleChangeBatch(List.of(event), committer) + ); + + assertInstanceOf(TimeoutException.class, failure.getCause()); + verify(firstHungSend).get(EXPECTED_ACKNOWLEDGEMENT_WAIT_MS, TimeUnit.MILLISECONDS); + verify(secondHungSend).get(EXPECTED_ACKNOWLEDGEMENT_WAIT_MS, TimeUnit.MILLISECONDS); + verify(kafkaTemplate, times(2)).send( + "orders.customer_updates", "customer-42", "{\"op\":\"u\"}" + ); + verify(committer, never()).markProcessed(any()); + verify(committer, never()).markBatchFinished(); + assertEquals(0L, cdcService.getStatus().get("kafkaPublishSuccess")); + assertEquals(2L, cdcService.getStatus().get("kafkaPublishFailure")); + } + + @SuppressWarnings("unchecked") + private static DebeziumEngine.RecordCommitter> committer() { + return (DebeziumEngine.RecordCommitter>) mock( + DebeziumEngine.RecordCommitter.class + ); + } + + @SuppressWarnings("unchecked") + private static ChangeEvent event() { + ChangeEvent event = (ChangeEvent) mock(ChangeEvent.class); + when(event.destination()).thenReturn("orders.customer_updates"); + when(event.key()).thenReturn("customer-42"); + when(event.value()).thenReturn("{\"op\":\"u\"}"); + return event; + } +} diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/service/CdcServiceDocumentationContractTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/service/CdcServiceDocumentationContractTest.java new file mode 100644 index 00000000..fca3ca9b --- /dev/null +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/service/CdcServiceDocumentationContractTest.java @@ -0,0 +1,64 @@ +package com.xtrmetl.cdc.service; + +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Guards beginner-readable documentation for the public CDC service lifecycle and construction + * surface that operators and embedded-module consumers call directly. + */ +class CdcServiceDocumentationContractTest { + + private static final Path CDC_SERVICE_SOURCE = projectRoot().resolve( + "cdc-service/src/main/java/com/xtrmetl/cdc/service/CdcService.java" + ); + + /** + * Requires every currently exposed construction and lifecycle entry point to explain its + * behavior instead of relying on method names or framework annotations alone. + * + * @throws IOException when the production source cannot be read + */ + @Test + void documentsPublicConstructionAndLifecycleSurface() throws IOException { + String source = Files.readString(CDC_SERVICE_SOURCE, StandardCharsets.UTF_8) + .replace("\r\n", "\n"); + + assertTrue(source.contains( + "Constructs the CDC service with explicit startup, canonical-mapping, and change-mapping dependencies." + )); + assertTrue(source.contains("Returns whether automatic CDC startup is enabled.")); + assertTrue(source.contains("Handles Spring Boot application readiness by applying the configured CDC auto-start policy.")); + assertTrue(source.contains("Releases CDC engine and executor resources when the Spring bean is destroyed.")); + } + + /** + * Locates the repository root for both reactor-root and module-local Maven execution. + * + * @return absolute repository root containing the root Maven project + */ + private static Path projectRoot() { + Path current = Paths.get(System.getProperty("user.dir")).toAbsolutePath(); + Path lastPomParent = null; + while (current != null) { + if (Files.exists(current.resolve(".git"))) { + return current; + } + if (Files.exists(current.resolve("pom.xml"))) { + lastPomParent = current; + } + current = current.getParent(); + } + if (lastPomParent != null) { + return lastPomParent; + } + throw new IllegalStateException("Could not find project root"); + } +} diff --git a/docs/cdc/ops-and-reliability.md b/docs/cdc/ops-and-reliability.md index 00a8fa7b..1bf0ef0e 100644 --- a/docs/cdc/ops-and-reliability.md +++ b/docs/cdc/ops-and-reliability.md @@ -15,6 +15,8 @@ Companion notes for `cdc-service`. Complements `ARCHITECTURE.md` and `README.md` |:--------------------|:--------|:-------------| | `CDC_AUTOSTART` / `xtrmetl.cdc.autostart` (or `mightyetl.cdc.autostart`) | Start engine on boot | `true` — ensure PG WAL slot capacity | | `KAFKA_BOOTSTRAP_SERVERS` | Event bus | Required for useful CDC | +| `CDC_KAFKA_DELIVERY_TIMEOUT_MS` | Kafka producer delivery success/failure upper bound | `60000` ms — lower values fail faster under broker pressure | +| `CDC_KAFKA_MAX_BLOCK_MS` | Bound producer metadata/buffer blocking | `30000` ms — lower values can surface transient pressure sooner | | `REPLICA_ENABLED` | Turn on Kafka→PG apply | `false` | | `REPLICA_TOPIC_PATTERN` | Topics to consume | `xtrmetl-cdc\..*` | | `REPLICA_TABLES` / `xtrmetl.replica.tables` | Tables for JDBC apply (`id`,`data` shape) | `processed_data` | @@ -24,13 +26,32 @@ Companion notes for `cdc-service`. Complements `ARCHITECTURE.md` and `README.md` Env helpers and validation live in `EnvUtils` / `ValidationUtils`. +## Source-to-Kafka acknowledgement boundary + +The live embedded-engine path uses Debezium's batch `ChangeConsumer` / `RecordCommitter` contract. For every destination-bearing change event, mightyETL submits the raw Debezium JSON to `KafkaTemplate`, waits interruptibly for the returned `CompletableFuture` to complete successfully for at most **65 seconds per application attempt**, and only then calls `RecordCommitter.markProcessed(...)`. `markBatchFinished()` is called only after the complete batch has reached its permitted processed state. + +Kafka publication uses a maximum of **two application attempts**. The first terminal send failure or 65-second future timeout increments `kafkaPublishFailure` and is retried once. A second terminal failure or timeout raises `KafkaException`; the failed record is not marked processed and the batch is not marked finished. Synchronous producer-send failures use the same bounded policy. Interruption while waiting for acknowledgement propagates without advancing the current Debezium record. + +The producer configuration fixes `acks=all`, enables producer idempotence, and limits `max.in.flight.requests.per.connection` to `5`. `CDC_KAFKA_DELIVERY_TIMEOUT_MS` and `CDC_KAFKA_MAX_BLOCK_MS` bound Kafka producer delivery reporting and producer blocking respectively. The application-level 65-second future wait is an independent guard: after `KafkaTemplate.send(...)` returns a future, mightyETL will not wait on that future indefinitely. The 65-second value deliberately exceeds the default Kafka delivery timeout of 60 seconds so normal producer terminal completion can surface first. Initiating `send(...)` can still be subject to producer blocking such as `max.block.ms`, so 65 seconds is a future-wait bound rather than a claim about total end-to-end attempt duration. + +`GET /api/cdc/status` exposes process-lifetime diagnostic counters: + +- `kafkaPublishSuccess`: acknowledged sends on the live batch path; +- `kafkaPublishFailure`: failed application send attempts, including timeouts and attempts later recovered by the one bounded retry. + +These counters contain no payloads, record keys, credentials, or exception text. + +This boundary is **at-least-once/replay-tolerant rather than an end-to-end exactly-once claim**. A crash can occur after Kafka acknowledgement but before Debezium's file-backed source offset is durably flushed, so a source event may be replayed after restart. Downstream consumers must retain stable event identity or idempotent processing where duplicate effects matter. See `docs/doctoring/cdc-kafka-acknowledged-delivery.md` for the failure model, TDD evidence, rollback boundary, and primary references. + ## Reliability features (real) +- Source publish: Kafka acknowledgement is observed before Debezium record progress; repeated terminal failure or acknowledgement timeout leaves the source record uncommitted. +- Source publish future wait: every returned Kafka send future is bounded to 65 seconds per application attempt; a non-terminating future cannot stall the CDC batch forever. - Replica consumer: Spring Kafka `DefaultErrorHandler` + **DLT** (`topic.DLT`) after retries. - `AckMode.RECORD` after successful apply. - Actuator: `health`, `info`, liveness/readiness probes enabled. - **CDC engine health:** component `cdcEngine` on `/actuator/health` (running / idle / down + slot details). -- Status API: replication slot lag bytes via `ReplicationSlotProbe`. +- Status API: replication slot lag bytes via `ReplicationSlotProbe` plus `kafkaPublishSuccess` / `kafkaPublishFailure` source-publish counters. - SPI: `postgres-debezium` start/stop delegates to `CdcService`. - Zipkin tracing sampling configurable via Spring Boot actuator. @@ -38,6 +59,7 @@ Env helpers and validation live in `EnvUtils` / `ValidationUtils`. | Gap | Impact | Mitigation | |:----|:-------|:-----------| +| Source→Kafka is not one distributed transaction | Crash after Kafka ack but before durable Debezium offset flush can replay a record | Keep downstream consumers idempotent/replay-tolerant; monitor publish failures; see acknowledged-delivery doctoring | | Slot lag is point-in-time probe | No historical lag graph | Scrape `/api/cdc/status` or use PG exporters | | Replica apply limited to `(id,data)` tables | Arbitrary schemas not replicated | Configure `replica.tables` only for matching shapes; warehouse connectors later | | One CDC process ownership | Duplicate processes fight for slots | Run single active instance per slot/publication | @@ -50,13 +72,13 @@ Env helpers and validation live in `EnvUtils` / `ValidationUtils`. | Method | Path | Purpose | |:-------|:-----|:--------| -| `GET` | `/api/cdc/status` | Engine flags, **replication slot lag** (`restartLagBytes` / `flushLagBytes`), replica flags, configured/registered sources & targets | +| `GET` | `/api/cdc/status` | Engine flags, Kafka publish counters, **replication slot lag** (`restartLagBytes` / `flushLagBytes`), replica flags, configured/registered sources & targets | | `GET` | `/api/cdc/sources` | Source SPI registry (Postgres Debezium today) | | `GET` | `/api/cdc/targets` | Target SPI registry (`kafka`, `jdbc-replica`) | | `POST` | `/api/cdc/start` | Start embedded Debezium engine | | `POST` | `/api/cdc/stop` | Stop engine | -Status JSON includes `product: mightyETL`, `anyToAny: false`, `replicationSlot`, and honest notes that capture is PostgreSQL→Kafka only. +Status JSON includes `product: mightyETL`, `anyToAny: false`, `kafkaPublishSuccess`, `kafkaPublishFailure`, `replicationSlot`, and honest notes that capture is PostgreSQL→Kafka only. ### replicationSlot fields @@ -77,7 +99,7 @@ psql -c "select slot_name, active, restart_lsn from pg_replication_slots;" curl -sf http://localhost:8001/actuator/health | jq . curl -sf http://localhost:8001/actuator/health/cdcEngine | jq . 2>/dev/null || true -# Engine / config status (no passwords) +# Engine / config status (no passwords); inspect Kafka publish counters as well curl -sf http://localhost:8001/api/cdc/status | jq . # Registered CDC source types @@ -88,10 +110,12 @@ curl -X POST http://localhost:8001/api/cdc/start curl -X POST http://localhost:8001/api/cdc/stop ``` +If `kafkaPublishFailure` rises, inspect broker reachability, topic authorization, producer/broker availability, configured delivery/block timeout bounds, and whether acknowledgement futures are timing out before restarting the engine. Do not treat a repeated retry as proof that the cause has been removed. + ## Tests -Unit tests under `cdc-service/src/test/java/com/xtrmetl/cdc/**` cover controller, service lifecycle mocks, replica appliers, Kafka error handler config, and **ops-doc alignment** (`ops/CdcOpsDocsAlignmentTest` — asserts this file names the same paths/health component the shipped `CdcController` / `CdcEngineHealthIndicator` expose). No Testcontainers integration suite in-repo yet. +Unit tests under `cdc-service/src/test/java/com/xtrmetl/cdc/**` cover controller, service lifecycle mocks, acknowledged Kafka publication before source progress, finite 65-second future-wait behavior, bounded terminal/synchronous publish failures, interruption, producer durability configuration, replica appliers, Kafka error handler config, and **ops-doc alignment** (`ops/CdcOpsDocsAlignmentTest` — asserts this file names the same paths/health component the shipped `CdcController` / `CdcEngineHealthIndicator` expose). No Testcontainers integration suite in-repo yet. ## Sale-ready honesty -Primary path is **PostgreSQL → Kafka** only. Warehouse BI connectors and multi-source any-to-any CDC remain scaffolds (see README support matrix and `docs/connectors/`). Operators should treat unlisted capabilities as **not production-supported**. +Primary path is **PostgreSQL → Kafka** only. Its embedded source publisher now waits for Kafka acknowledgement with a finite application wait before marking a Debezium record processed, but this does not make PostgreSQL, the file-backed Debezium offset store, Kafka, and downstream consumers one exactly-once transaction. Warehouse BI connectors and multi-source any-to-any CDC remain scaffolds (see README support matrix and `docs/connectors/`). Operators should treat unlisted capabilities as **not production-supported**. diff --git a/docs/doctoring/cdc-kafka-acknowledged-delivery.md b/docs/doctoring/cdc-kafka-acknowledged-delivery.md new file mode 100644 index 00000000..c702bd25 --- /dev/null +++ b/docs/doctoring/cdc-kafka-acknowledged-delivery.md @@ -0,0 +1,144 @@ +# PostgreSQL-to-Kafka acknowledged-delivery evidence + +Reviewed on: **2026-08-08** + +## Buyer-visible reliability gap + +The embedded PostgreSQL CDC path previously registered a one-record Debezium consumer and returned immediately after `KafkaTemplate.send(...)`. Spring Kafka returns a `CompletableFuture>` from the send operation, so returning from the handler proved only that a send was submitted to the producer API; the repository had no executable contract that Kafka had acknowledged that record before Debezium was allowed to advance processing state. + +For an ETL/CDC product, that boundary is commercially material. A source position that advances independently of target acknowledgement can turn an ordinary broker outage or producer failure into an avoidable delivery ambiguity. The repair must therefore bind the source-processing decision to observable target completion without claiming a stronger end-to-end exactly-once guarantee than the architecture actually provides. + +A second reliability boundary became visible after acknowledgement waiting was implemented: the application originally used an unbounded `CompletableFuture.get()`. Kafka producer delivery and blocking limits are important lower-layer controls, but a provider defect or future that never reaches a terminal state could still leave the application-level Debezium batch waiting indefinitely. The application now enforces its own finite acknowledgement wait for every publication attempt. + +## Root-cause analysis + +The original defect was a mismatch between two asynchronous contracts: + +1. Spring Kafka publishes through a future-returning `KafkaTemplate.send(...)` API. The caller must observe future completion when downstream acknowledgement is part of the caller's correctness condition. +2. Debezium Engine exposes the advanced `ChangeConsumer` / `RecordCommitter` contract so an embedded consumer can mark individual records processed and mark a batch finished only after application processing has completed. + +The old path used neither completion signal together. It submitted the Kafka operation from a simple consumer and then returned. Retrying the whole workflow, increasing broker timeouts, or merely setting stronger producer acknowledgement options would not repair that application-level ordering defect by themselves. + +The follow-up timeout defect was at the application/future boundary: `awaitAcknowledgement(...)` called the no-timeout `CompletableFuture.get()`. Producer `delivery.timeout.ms` bounds Kafka's delivery reporting after a record has been accepted by the producer and `max.block.ms` bounds specific blocking producer operations, but neither substitutes for an explicit bound on this application's future wait. The smallest direct repair is therefore a timed future wait that remains interruptible and feeds the existing retry/fail-closed path. + +## Feasibility analysis + +The following remediation classes were evaluated against the current repository boundary. + +### Executable now + +Use Debezium's batch `ChangeConsumer` contract, await each Spring Kafka send future, and call `RecordCommitter.markProcessed(...)` only after successful completion. Mark the batch finished only after every record has reached its permitted terminal state. Keep the behavior in the existing `cdc-service` so no new service, credential, database, or separately leased repository is required. + +Bound each application-level future wait to 65 seconds. This deliberately exceeds the repository's default Kafka `delivery.timeout.ms` of 60 seconds so normal producer terminal completion can surface first, while still giving the application a finite fail-closed boundary if a future itself never completes. A timeout is treated as a failed publication attempt and therefore follows the existing one-retry policy. No new secret, provider, endpoint, or cross-repository authority is needed. + +These are the selected remediations because they address the causes directly and can be tested deterministically with controlled `CompletableFuture` completion and timeout behavior. + +### Complementary producer controls + +Require `acks=all`, explicit producer idempotence, `max.in.flight.requests.per.connection=5`, a 60-second default `delivery.timeout.ms`, and a 30-second default `max.block.ms`. These controls strengthen Kafka producer delivery behavior and bound lower-layer producer stalls, but they do not replace either the application-level acknowledgement-before-offset contract or the application's finite future-wait boundary. + +### Not selected as this bounded slice + +Migrating the product to Kafka Connect distributed mode, adding a transactional outbox, introducing an external offset/acknowledgement ledger, or adding a new timeout configuration surface could provide different durability or tuning properties, but each is a materially larger architecture/configuration change. None is necessary to correct these specific embedded-engine ordering and unbounded-wait defects, and none should be introduced merely to make this pull request appear stronger. + +## Implemented delivery state machine + +For each Debezium event in a batch: + +```text +receive ordered change event +→ if destination exists, run the optional canonical mapping observation +→ submit raw Debezium JSON to Kafka +→ wait interruptibly for acknowledgement for at most 65 seconds +→ on acknowledgement: increment kafkaPublishSuccess +→ mark this Debezium record processed +→ continue to the next event +→ after every record is processed: mark the Debezium batch finished +``` + +A destination-less engine event is treated as non-publishable metadata and is marked processed without touching Kafka or the publish counters. + +A failed or timed-out Kafka attempt increments `kafkaPublishFailure` and receives one bounded application-level retry. A second terminal failure or timeout raises `KafkaException`; the current record is not marked processed and the batch is not marked finished. A synchronous exception raised while initiating the send follows the same bounded retry policy. An `InterruptedException` while awaiting acknowledgement propagates immediately and does not advance the current record or batch. + +The live Debezium engine is wired to this batch handler. The older one-record `handleChangeEvent(...)` method remains as a compatibility surface for direct callers and tests, but it is no longer the live source-offset progression path. + +## Producer durability controls + +`cdc-service/src/main/resources/application.yml` now requires: + +- `acks: all`; +- `enable.idempotence=true`; +- `max.in.flight.requests.per.connection=5`; +- `delivery.timeout.ms=${CDC_KAFKA_DELIVERY_TIMEOUT_MS:60000}`; +- `max.block.ms=${CDC_KAFKA_MAX_BLOCK_MS:30000}`. + +Apache Kafka documents that producer idempotence requires `acks=all`, retries greater than zero, and no more than five in-flight requests per connection. Kafka also defines `delivery.timeout.ms` as the upper bound for reporting success or failure after a record is accepted by the producer. The application leaves Kafka's retry machinery intact, adds one bounded retry after a terminal send-future failure, and independently caps each future acknowledgement wait at 65 seconds. + +The 65-second application wait is not a claim that a complete attempt can consume only 65 seconds: initiating `KafkaTemplate.send(...)` can itself be subject to producer blocking behavior such as `max.block.ms`. The important invariant is narrower and testable: after a send future has been returned, this application will not wait on that future forever. + +## Operator evidence + +`GET /api/cdc/status` now includes cumulative process-lifetime counters: + +- `kafkaPublishSuccess`: records whose Kafka send completed successfully through the acknowledged live path; +- `kafkaPublishFailure`: failed application send attempts, including timeout failures and failures that were subsequently recovered by the one bounded retry. + +These counters intentionally contain no payload, source key, principal, topic contents, credential, or exception text. They are diagnostic evidence, not a durable billing or audit ledger. + +## TDD evidence + +### RED — missing acknowledgement boundary + +Commit `f7ca5f149df959d03ff330ea0e374bc8fcb031e4` introduced the initial contract tests before production implementation. CI run `31259345232` failed because `CdcService` did not provide `handleChangeBatch(...)`. That failure demonstrates that the requested acknowledgement/committer contract did not already exist. + +### Strengthened RED — terminal and synchronous failure behavior + +Commit `b378c98c1ef50f26e7d5d321dc924ef854010b81` added bounded terminal-failure and synchronous-send-failure cases before implementation. CI run `31261230985`, including macOS job `93112224276`, failed at test compilation because all new batch-contract calls still targeted the missing production method. The existing production suite was not rewritten to manufacture a passing result. + +### GREEN implementation + +Commit `15153d038edb757189e56de6c748193a5d03242f` added the acknowledged batch handler, counters, bounded retry, interrupt propagation, and live Debezium `ChangeConsumer` wiring. Commit `254ea3fb241528392a8196db8b82f85ffea91f6d` added the Kafka producer controls. A subsequent test-compilation failure revealed only that one Mockito verification method needed to declare the checked `InterruptedException` from Debezium's `RecordCommitter`; commit `1ae9d757fd9f35859f51b1f3ce750d6b7bac443c` corrected that test signature without weakening an assertion. + +On that implementation head, macOS CI job `93112811576` ran the full Maven reactor successfully. `CdcKafkaPublishAcknowledgementTest` ran **8 tests with 0 failures, 0 errors, and 0 skips**, and the complete `cdc-service` suite ran **114 tests with 0 failures, 0 errors, and 0 skips**. This is development evidence only; every later documentation commit invalidates it as exact-current-head merge evidence and must receive its own checks. + +### RED — unbounded future wait + +Commit `85e9f8a3369e39634b57295c24f52c9e89bf5917` added `CdcKafkaPublishTimeoutTest` before changing production. PR-triggered CI run `31276261918`, Ubuntu job `93150176837`, compiled the production CDC service and all tests successfully and then ran the new boundary test. The CDC module ran **116 tests with exactly 1 failure and 0 errors**: `timesOutHungKafkaAcknowledgementsWithoutAdvancingOffsets` expected the two mocked futures' timed `get(65000, MILLISECONDS)` calls to fail closed, but production still used the unbounded no-argument `get()` and therefore returned without throwing. Existing `CdcKafkaPublishAcknowledgementTest` remained **8/8 green**. The failure reached the intended production/future boundary and was not a fixture, import, compilation, or environment defect. + +### GREEN — finite future wait + +Commit `c56dce4586ca330e7b199a2651010eb50283b5e5` changed only the production acknowledgement boundary: each returned send future is now awaited interruptibly for at most 65 seconds; `TimeoutException` is wrapped into the existing `ExecutionException` failure path so retry counters, terminal `KafkaException`, and offset fail-closed behavior remain unchanged. On PR-triggered CI run `31276397082`, Ubuntu job `93150509726` ran the full Maven reactor successfully. `CdcKafkaPublishTimeoutTest` passed **1/1**, `CdcKafkaPublishAcknowledgementTest` passed **8/8**, and the complete CDC module passed **116 tests with 0 failures, 0 errors, and 0 skips**. Because the protected `develop` workflow still checks out a synthetic PR merge, this is valid development/TDD proof but is not being misrepresented as literal-head merge evidence. + +## Failure and recovery behavior + +If Kafka does not acknowledge a destination-bearing change after both application attempts, including a future that fails to complete within 65 seconds on an attempt, the handler fails closed for source progress. Debezium may later replay source records according to its offset-storage and restart semantics. Operators should therefore investigate broker availability, topic authorization, producer configuration, network reachability, and future/provider health before restarting or repeatedly retrying the service. + +`CDC_KAFKA_DELIVERY_TIMEOUT_MS` and `CDC_KAFKA_MAX_BLOCK_MS` are lower-layer operational bounds, not success guarantees. Lowering them aggressively can increase terminal producer failures during transient pressure. Raising them increases lower-layer wait time. Independently, the application will wait at most 65 seconds on each returned send future before treating that attempt as failed; this prevents a non-terminating future from stalling the CDC batch indefinitely. + +## Exactly-once limitation + +This change **does not claim end-to-end exactly-once delivery**. + +Kafka producer idempotence protects producer retry behavior within Kafka's documented producer semantics. It does not make the external PostgreSQL source position, this process's file-backed Debezium offset store, Kafka acknowledgement, and every downstream consumer one distributed transaction. A process crash can occur after Kafka acknowledgement but before a durable Debezium offset flush, allowing the source event to be replayed after restart. Application-level retry after a terminal send failure or timeout can also encounter an ambiguous prior outcome. + +Downstream consumers must therefore remain replay-tolerant and use stable event identity or idempotent application semantics where duplicate effects matter. A future transactional-outbox or externally coordinated delivery design would require its own failure model and evidence. + +## Security and privacy properties + +- No new credential or secret is introduced. +- No payload, record key, principal, or exception message is added to the operator status response. +- No protected branch, repository policy, or separately leased repository is modified by this slice. +- The Kafka producer remains configured through existing Spring Boot configuration and existing runtime credentials/network boundaries. +- Failed or timed-out publication prevents source progress rather than converting an unavailable sink into a successful offset transition. + +## Rollback + +A rollback must remove the batch `RecordCommitter` implementation, producer configuration, finite application acknowledgement wait, status counters, tests, operations guidance, architecture description, and this evidence together only after an equivalent or stronger acknowledgement-before-source-progress mechanism is available. Do not roll back by restoring fire-and-forget or unbounded Kafka publication while retaining documentation that claims acknowledgement-bound, bounded source progress. + +## References + +Apache Software Foundation. (2025). *Producer configs*. Apache Kafka 3.9 documentation. https://kafka.apache.org/39/configuration/producer-configs/ + +Debezium. (2026). *Debezium Engine*. Debezium documentation. https://debezium.io/documentation/reference/development/engine.html + +Spring. (n.d.). *Sending messages*. Spring for Apache Kafka reference documentation. Retrieved August 8, 2026, from https://docs.spring.io/spring-kafka/reference/kafka/sending-messages.html