Designing Replay-Safe CDC Pipelines With Kafka, Debezium, and Recovery Contracts

How to design CDC pipelines with Kafka, Debezium, idempotent writes, deterministic projections, replay workflows, reconciliation checks, and recovery evidence.

Ishan Shah

Sep. 01, 26 · Tutorial

Likes (0)

Comment

Save

140 Views

Change data capture (CDC) pipelines look straightforward on paper: capture database changes, publish them to Kafka, and update downstream systems. The difficulty starts when events are duplicated, consumers restart, projections drift, or a team needs to replay months of history without corrupting the state it is trying to recover.

A reliable CDC design has to account for those failure modes from the beginning. That means combining Kafka and Debezium with idempotent writes, deterministic projections, controlled replay workflows, reconciliation checks, and enough recovery evidence to explain what happened when something goes wrong.

The architecture:

The goal is not only to move inventory changes quickly. The goal is to make replay safe enough that operators can rebuild and explain the derived state after failure.

This article builds one concrete pattern:

The important detail is that replay safety is not a single feature. It is the result of several boring decisions lining up correctly.

Data Model

The data model should separate the aggregate state, the classification state, and the transaction history.

    PLSQL
   
   CREATE TABLE inventory_stock_on_hand (

  sku             VARCHAR(64) PRIMARY KEY,

  stock_on_hand   BIGINT NOT NULL,

  updated_at      TIMESTAMP NOT NULL

);

CREATE TABLE inventory_bucket (

  sku             VARCHAR(64) NOT NULL,

  bucket_type     VARCHAR(32) NOT NULL,

  location_id     VARCHAR(64) NOT NULL,

  quantity        BIGINT NOT NULL,

  updated_at      TIMESTAMP NOT NULL,

  PRIMARY KEY (sku, bucket_type, location_id)

);

CREATE TABLE inventory_transaction (

  event_id        VARCHAR(128) PRIMARY KEY,

  sku             VARCHAR(64) NOT NULL,

  seller_id       VARCHAR(64) NOT NULL,

  delta_quantity  BIGINT NOT NULL,

  event_time      TIMESTAMP NOT NULL,

  accepted_at     TIMESTAMP NOT NULL

);

CREATE INDEX idx_inventory_transaction_sku_time

  ON inventory_transaction (sku, event_time);

CREATE INDEX idx_inventory_bucket_sku_bucket

  ON inventory_bucket (sku, bucket_type);

The transaction table is the recovery anchor. If the availability projection drifts, the system needs a history to explain the projection.

Do not rely only on the mutable aggregate table. inventory_stock_on_hand is useful for fast reads, but it is not enough for recovery. If the aggregate is wrong, it cannot explain how it became wrong. The accepted transaction history gives replay something durable to reason from.

Ingestion Event

Use an event ID that can survive retries and replay.

    JSON
   
   {

  "event_id": "mkt-evt-8f11a",

  "sku": "1231241",

  "quantity": 100,

  "operation": "I",

  "event_time": "2026-06-19T18:23:11Z",

  "seller_id": "seller-42"

}

The consumer should perform an idempotent write. One pattern is to insert the transaction first using event_id as the primary key. If the insert fails because the event already exists, skip the duplicate and emit a duplicate-suppression metric.

    Java
   
   public InventoryWriteResult apply(InventoryEvent event) {

    try {

        transactionRepository.insert(event.toTransactionRow());

    } catch (DuplicateKeyException duplicate) {

        metrics.increment("inventory.duplicate_event");

        return InventoryWriteResult.duplicate(event.eventId());

    }

    stockRepository.incrementStockOnHand(event.sku(), event.quantity());

    bucketRepository.incrementBucket(event.sku(), "SELLABLE", event.quantity());

    return InventoryWriteResult.accepted(event.eventId());

}

In production, the accepted transaction insert and the aggregate updates should be part of the same database transaction. A useful shape is:

    PLSQL
   
   BEGIN;

WITH accepted AS (

  INSERT INTO inventory_transaction (

    event_id,

    sku,

    seller_id,

    delta_quantity,

    event_time,

    accepted_at

  ) VALUES (

    :event_id,

    :sku,

    :seller_id,

    :delta_quantity,

    :event_time,

    now()

  )

  ON CONFLICT (event_id) DO NOTHING

  RETURNING sku, delta_quantity

)

INSERT INTO inventory_stock_on_hand (sku, stock_on_hand, updated_at)

SELECT sku, delta_quantity, now()

FROM accepted

ON CONFLICT (sku) DO UPDATE

SET stock_on_hand = inventory_stock_on_hand.stock_on_hand + EXCLUDED.stock_on_hand,

    updated_at = now();

COMMIT;

That ON CONFLICT clause is not just a database convenience. It is part of the replay contract. It ensures that retrying the same business event does not apply the same inventory delta twice.

Debezium Configuration

Enable PostgreSQL logical decoding and configure Debezium to emit CDC topics for the inventory tables.

    JSON
   
   {

  "name": "postgres-inventory-connector",

  "config": {

    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",

    "database.hostname": "<POSTGRES_HOSTNAME>",

    "database.port": "5432",

    "database.user": "<POSTGRES_USER>",

    "database.password": "<POSTGRES_PASSWORD>",

    "database.dbname": "<POSTGRES_DBNAME>",

    "topic.prefix": "inventory_source",

    "plugin.name": "pgoutput",

    "slot.name": "debezium_inventory_slot",

    "publication.autocreate.mode": "filtered",

    "table.include.list": "public.inventory_stock_on_hand,public.inventory_bucket,public.inventory_transaction",

    "snapshot.mode": "initial",

    "heartbeat.interval.ms": "10000",

    "tombstones.on.delete": "false",

    "key.converter": "org.apache.kafka.connect.json.JsonConverter",

    "value.converter": "org.apache.kafka.connect.json.JsonConverter",

    "key.converter.schemas.enable": "true",

    "value.converter.schemas.enable": "true"

  }

}

Debezium gives you history, but not recovery confidence. The confidence comes from how you key, project, replay, and reconcile that history.

For replay work, track these connector facts in your runbook:

Connector name and version
Replication slot name
Publication name and included tables
Snapshot mode used for initial load
Topic prefix
Last processed LSN
Connector lag
Schema history topic

When a connector interruption happens, those details tell you whether you can resume normally, need a bounded replay, or need a new snapshot plus downstream reconciliation.

Partition-Aware Routing

The partition key should be chosen from the business ordering boundary.

    Java
   
   public class SkuPartitioner implements Partitioner {

    @Override

    public int partition(

            String topic,

            Object key,

            byte[] keyBytes,

            Object value,

            byte[] valueBytes,

            Cluster cluster) {

        InventoryEvent event = (InventoryEvent) value;

        String orderingKey = event.getSku();

        int partitionCount = cluster.partitionCountForTopic(topic);

        return Math.floorMod(orderingKey.hashCode(), partitionCount);

    }

}

Partitioning is not merely a throughput setting. If the projection depends on entity-local ordering, the entity belongs in the key.

Kafka Streams Topology

A simplified topology might rekey CDC records by SKU, materialize source tables, and compute availability.

    Java
   
   StreamsBuilder builder = new StreamsBuilder();

KTable<String, StockOnHand> stock =

    builder.table("inventory_source.public.inventory_stock_on_hand",

        Consumed.with(Serdes.String(), stockSerde));

KTable<String, InventoryBuckets> buckets =

    builder.table("inventory_source.public.inventory_bucket",

        Consumed.with(Serdes.String(), bucketSerde));

KTable<String, AvailabilityProjection> availability =

    stock.join(

        buckets,

        (stockRow, bucketRows) -> AvailabilityProjection.compute(stockRow, bucketRows),

        Materialized.<String, AvailabilityProjection, KeyValueStore<Bytes, byte[]>>as("availability-store")

            .withKeySerde(Serdes.String())

            .withValueSerde(availabilitySerde)

    );

availability

    .toStream()

    .filter((sku, projection) -> projection.isPublishable())

    .to("inventory.availability.v2", Produced.with(Serdes.String(), availabilitySerde));

The projection function should be deterministic.

If replaying the same accepted history does not produce the same projection, the topology is not replay-safe.

Recovery Contract

Attach a Recovery Contract to the flow.

    YAML
   
   recovery_contract:

  flow: inventory-availability-projection

  tuple: "<H, O, I, F, S, Q, E>"

  history:

    source:

      - inventory_transaction

      - debezium.inventory_transaction

  order:

    key: sku

  idempotency:

    key: event_id

    duplicate_policy: skip_and_report

  function:

    name: compute_sellable_availability

    deterministic: true

  scope:

    supported:

      - by_sku

      - by_time_window

      - by_partition

  checks:

    - stock_on_hand_matches_transactions

    - sellable_quantity_non_negative

    - projection_event_time_valid

  evidence:

    - replay_scope

    - events_processed

    - duplicates_skipped

    - projections_changed

    - reconciliation_failures

    - confidence_status

Treat this file as executable architecture documentation. A service should fail fast if the contract is incomplete for a critical flow.

    Java
   
   public final class RecoveryContractValidator {

    public void validate(RecoveryContract contract) {

        requireNonEmpty(contract.flow(), "flow");

        requireNonEmpty(contract.history().source(), "history.source");

        requireNonEmpty(contract.order().key(), "order.key");

        requireNonEmpty(contract.idempotency().key(), "idempotency.key");

        requireNonEmpty(contract.function().name(), "function.name");

        requireTrue(contract.function().deterministic(), "projection must be deterministic");

        requireNonEmpty(contract.scope().supported(), "scope.supported");

        requireNonEmpty(contract.checks(), "checks");

        requireNonEmpty(contract.evidence(), "evidence");

    }

    private void requireNonEmpty(Object value, String field) {

        if (value == null || value.toString().isBlank()) {

            throw new IllegalArgumentException("Missing recovery contract field: " + field);

        }

    }

    private void requireTrue(boolean value, String message) {

        if (!value) {

            throw new IllegalArgumentException(message);

        }

    }

}

That validator does not make the system correct by itself. It prevents a more common failure: discovering during an incident that nobody defined the replay scope, idempotency key, or reconciliation checks.

Replay Workflow

Replay should be treated as a controlled workflow.

    Plain Text
   
Identify incident scope.

Select replay scope by SKU, time window, or partition.

Read authoritative history.

Rebuild deterministic projection.

Run reconciliation checks.

Emit recovery evidence.

Republish only if checks pass.

The output should be an evidence report.

    JSON
   
{

  "recovery_id": "rec-2026-06-19-001",

  "flow": "inventory-availability-projection",

  "events_processed": 1842,

  "duplicates_skipped": 17,

  "projection_rows_changed": 11,

  "reconciliation": {

    "stock_on_hand_matches_transactions": true,

    "sellable_quantity_non_negative": true,

    "projection_event_time_valid": true

  },

  "confidence_status": "trusted"

}

A replay runner can keep the workflow explicit:

    Java
   
   public RecoveryEvidence replay(ReplayRequest request) {

    RecoveryContract contract = contracts.load(request.flow());

    validator.validate(contract);

    ReplayScope scope = scopeResolver.resolve(request, contract);

    List<InventoryEvent> history = historyReader.read(contract.history(), scope);

    ReplayResult result = projector.rebuild(history, contract.function());

    ReconciliationResult reconciliation =

        reconciliationRunner.run(contract.checks(), scope, result);

    RecoveryEvidence evidence = RecoveryEvidence.builder()

        .recoveryId(UUID.randomUUID().toString())

        .flow(request.flow())

        .scope(scope)

        .eventsProcessed(history.size())

        .duplicatesSkipped(result.duplicatesSkipped())

        .projectionsChanged(result.changedRows())

        .reconciliation(reconciliation)

        .confidenceStatus(reconciliation.passed() ? "trusted" : "review_required")

        .build();

    evidenceStore.write(evidence);

    if (request.publish() && reconciliation.passed()) {

        publisher.publish(result.projections());

    }

    return evidence;

}

The replay runner should support dry runs. Dry runs let operators answer "What would change?" before republishing availability, billing, or detection outputs.

Operational Metrics

Track ordinary health and recovery confidence separately.

Ordinary health:

Consumer lag
Connector lag
Task restarts
DLQ count
End-to-end latency

Recovery confidence:

Replay duration
Replay scope size
Duplicate suppression count
Projection rows changed
Reconciliation failures
Confidence status

Example metric names:

    Plain Text
   
   inventory_ingest_events_total{result="accepted|duplicate|rejected"}

inventory_cdc_connector_lag_seconds{connector="postgres-inventory-connector"}

inventory_stream_projection_lag_seconds{topology="availability"}

inventory_replay_duration_seconds{flow="inventory-availability-projection"}

inventory_replay_events_processed_total{flow="inventory-availability-projection"}

inventory_replay_duplicates_skipped_total{flow="inventory-availability-projection"}

inventory_reconciliation_failures_total{check="stock_on_hand_matches_transactions"}

inventory_recovery_confidence_status{status="trusted|review_required|failed"}

Alert on disagreement, not only lag. A good pipeline can be caught up and still be wrong.

    YAML
   
   alerts:

  - name: InventoryProjectionReconciliationFailure

    expr: inventory_reconciliation_failures_total > 0

    severity: page

  - name: InventoryReplayRequiresReview

    expr: inventory_recovery_confidence_status{status="review_required"} > 0

    severity: ticket

  - name: InventoryConnectorLagHigh

    expr: inventory_cdc_connector_lag_seconds > 300

    severity: ticket

Reconciliation Queries

Reconciliation should be executable, not just a diagram in a runbook. Start with invariants that are simple enough to automate.

Example: Stock-on-hand should match accepted transaction deltas for a replay window.

    PLSQL
   
   WITH accepted_delta AS (

  SELECT

    sku,

    SUM(delta_quantity) AS expected_delta

  FROM inventory_transaction

  WHERE accepted_at BETWEEN :from_time AND :to_time

  GROUP BY sku

),

actual_delta AS (

  SELECT

    sku,

    stock_on_hand - :baseline_stock_on_hand AS observed_delta

  FROM inventory_stock_on_hand

  WHERE sku = :sku

)

SELECT

  a.sku,

  a.expected_delta,

  b.observed_delta,

  (a.expected_delta = b.observed_delta) AS matches

FROM accepted_delta a

JOIN actual_delta b ON a.sku = b.sku;

Example: Sellable inventory should never be negative.

    PLSQL
   
   SELECT sku, location_id, quantity

FROM inventory_bucket

WHERE bucket_type = 'SELLABLE'

  AND quantity < 0;

These queries are not academically exciting, but they are operationally powerful. They turn "the replay finished" into "the replay finished and the invariants passed."

Replay Endpoint Sketch

A replay workflow should be explicit and permissioned. One possible internal API:

    HTTP
   
POST /internal/recovery/replay

Content-Type: application/json

{

  "flow": "inventory-availability-projection",

  "scope": {

    "type": "sku_and_time_window",

    "sku": "1231241",

    "from_event_time": "2026-06-19T18:00:00Z",

    "to_event_time": "2026-06-19T19:00:00Z"

  },

  "dry_run": false,

  "requested_by": "sre-oncall",

  "reason": "projection drift after stream task restart"

}

The response should not just say 200 OK.

    JSON
   
   {

  "recovery_id": "rec-2026-06-19-001",

  "status": "trusted",

  "events_processed": 1842,

  "duplicates_skipped": 17,

  "projections_changed": 11,

  "reconciliation_failures": 0,

  "evidence_uri": "<RECOVERY_EVIDENCE_URI>"

}

The response is the operational artifact. It gives the team something to attach to an incident timeline and something to compare against later recovery runs.

Tests for Replay Safety

Replay safety should be tested before production incidents.

    Java
   
   @Test

void replayingSameHistoryDoesNotChangeProjectionTwice() {

    List<InventoryEvent> history = List.of(

        event("evt-1", "SKU-1", 10),

        event("evt-2", "SKU-1", -2),

        event("evt-1", "SKU-1", 10) // duplicate

    );

    AvailabilityProjection first = projector.replay(history);

    AvailabilityProjection second = projector.replay(history);

    assertThat(first).isEqualTo(second);

    assertThat(first.sellableQuantity()).isEqualTo(8);

    assertThat(first.duplicatesSkipped()).isEqualTo(1);

}

Also test late events, schema versions, partition rebalance, connector restart, and partial replay by entity. If replay is part of your recovery model, it deserves the same test discipline as the happy-path pipeline.

Add failure injection tests that mirror production recovery:

    Java
   
   @Test

void lateEventTriggersReviewWhenItChangesPublishedAvailability() {

    ReplayScope scope = ReplayScope.forSkuAndWindow(

        "SKU-1",

        Instant.parse("2026-06-19T18:00:00Z"),

        Instant.parse("2026-06-19T19:00:00Z")

    );

    history.append(event("evt-1", "SKU-1", 10, "2026-06-19T18:01:00Z"));

    history.append(event("evt-2", "SKU-1", -3, "2026-06-19T18:59:00Z"));

    history.appendLate(event("evt-3", "SKU-1", -2, "2026-06-19T18:30:00Z"));

    RecoveryEvidence evidence = replayRunner.replay(

        ReplayRequest.dryRun("inventory-availability-projection", scope)

    );

    assertThat(evidence.eventsProcessed()).isEqualTo(3);

    assertThat(evidence.projectionsChanged()).isGreaterThan(0);

    assertThat(evidence.confidenceStatus()).isEqualTo("review_required");

}

Failure Injection Matrix

Use a small matrix before every major release of the pipeline.

Duplicate Event
- Injection: Send the same event_id twice.
- Expected evidence: duplicates_skipped > 0; no double-counted stock.
Late Event
- Injection: Delay event arrival until after the projection has already published output.
- Expected evidence: late event count, changed projections, and review status if the output changes.
Connector Pause
- Injection: Stop the Debezium connector for several minutes.
- Expected evidence: connector lag, replay scope, and reconciliation status.
Offset Rewind
- Injection: Reprocess a known event range.
- Expected evidence: deterministic replay agreement.
Schema Change
- Injection: Replay old and new schema versions.
- Expected evidence: schema versions recorded in the recovery evidence.
Bad projection deploy
- Injection: Publish an incorrect derived state, then replay.
- Expected evidence: projections changed; reconciliation passes after rebuild.

The point is not to create chaos for its own sake. The point is to practice the exact recovery motion before a real incident.

Production Hardening Checklist

Before relying on replay in production, confirm:

The authoritative history has retention longer than the largest expected recovery window.
The idempotency key is stable across producer retries.
The Kafka partition key matches the business ordering boundary.
The projection function is deterministic for the supported replay scope.
The contract names every source topic, source table, check, and evidence field.
The replay endpoint supports dry runs.
Republish requires reconciliation success.
Evidence is written to durable storage.
Evidence records include schema versions and replay input bounds.
Operators can find the runbook from the alert.
The DLQ is treated as an input to recovery, not as the recovery plan itself.

For high-value flows, make this checklist part of the architecture review. It is much cheaper to define replay semantics while designing the pipeline than to invent them under pressure.

Common Mistakes

Treating CDC topics as transient integration messages instead of durable recovery history.
Choosing partition keys for infrastructure convenience rather than business ordering.
Allowing stream processors to perform hidden non-idempotent side effects.
Measuring lag but not correctness.
Resetting offsets without a reconciliation plan.
Assuming exactly-once semantics removes the need for recovery evidence.

Conclusion

Replay-safe CDC pipelines require more than Kafka, Debezium, and stream processing. They require explicit recovery semantics. Recovery Contracts give teams a compact way to define those semantics. Confidence-carrying replay gives operators evidence that the recovered state can be trusted. That is the difference between a pipeline that resumes and a platform that actually recovers.

kafka Pipeline (software) Data Types

Opinions expressed by DZone contributors are their own.