KDS CDC Ingestion Solution

Purpose

This document describes the chosen production approach for ingesting Logic Control KDS CDC data into KDSAI warehouse storage.

Related notes:

• Performance And Scaling Notes
• Debezium Payload Compatibility Notes

The chosen approach is a framework-native SQS micro-batch CDC adapter. It follows the same ETL stage model used by the existing Brink POS adapter while replacing source-specific extraction with Debezium CDC messages delivered through cross-account SQS FIFO.

The design goals are:

• preserve the current ETL adapter abstraction,
• keep the existing S3 company/tenant key structure,
• avoid one ingestion batch per CDC record,
• persist accepted CDC data in S3 raw for replay,
• load Redshift through S3 Parquet and COPY,
• quarantine bad records without blocking good records,
• support global CDC ordering,
• include curated-to-S3 export from the start.

Selected Behavior Summary

Integration Flow

flowchart TD
    A[Logic Control KDS DB] --> B[Debezium connector<br/>source AWS account]
    B --> C[SQS FIFO queue<br/>cross-account]
    C --> D[KDS CDC background worker]
    D --> E[Poll and validate messages]
    E --> F[Resolve source customer/account<br/>and store_guid]
    F --> G{Mapping valid?}
    G -- no --> Q[Quarantine bad record<br/>with raw/source diagnostics]
    G -- yes --> H[Build ordered micro-batch<br/>500-2,000 records or age limit]
    H --> I[extract_raw<br/>write S3 raw by company/tenant/entity]
    I --> J[Create durable internal job/batch]
    J --> K[Delete SQS messages]
    J --> L[normalize_load<br/>write S3 load Parquet]
    L --> M[load_to_staging<br/>Redshift COPY]
    M --> N[staging_to_curated<br/>soft-delete/upsert]
    N --> O[curated_to_s3<br/>Redshift UNLOAD]
    O --> P[S3 curated Parquet]

Stage 1: Debezium To SQS FIFO

Debezium runs in the source AWS account against the Logic Control KDS database. It emits CDC envelopes to SQS FIFO.

The KDSAI side treats SQS as the delivery buffer, not the long-term replay system. SQS messages are temporary transport records. S3 raw is the replay source after acknowledgement.

Expected message content:

• source database/schema/table,
• source operation code,
• source primary key,
• source customer/account identifier,
• source store_guid where available,
• source is_deleted where available,
• Debezium source coordinates or offset metadata,
• SQS FIFO message group and sequence metadata.

Global order is required. The worker must use FIFO sequence metadata and Debezium source coordinates to detect gaps, duplicates, and out-of-order messages. If global order cannot be proven for a batch, the batch must fail safely before Redshift load.

Stage 2: Worker Pull And Micro-Batch Creation

The KDS CDC background worker polls SQS FIFO and builds micro-batches.

Batch close triggers:

The worker must not create one ingestion batch per CDC record. One ETL batch represents a bounded set of ordered CDC records for a tenant/entity slice.

Bad records are isolated during pull/validation when possible. A malformed record or unmapped tenant record is quarantined with diagnostics, while good records continue into the batch.

Stage 3: Tenant Resolution

Tenant resolution happens before any tenant-scoped S3 path or Redshift load is created.

Mapping rule:

source customer/account -> company_id
source store_guid -> tenant_id within company_id
company_id + tenant_id -> tenant Redshift database

The source field names do not need to be named company_id or tenant_id. They must map explicitly to platform values through a registered mapping table or mapping service.

No implicit tenant inference is allowed. If mapping is missing or ambiguous, the record is quarantined and not loaded to Redshift.

Required quarantine metadata for mapping failures:

• source customer/account value,
• source store_guid,
• source table,
• source primary key,
• Debezium source coordinates,
• SQS message id and sequence metadata,
• raw S3 path when available,
• reason mapping failed.

Stage 4: Store To S3 Raw (extract_raw)

extract_raw persists accepted CDC records to S3 raw.

Raw storage keeps the existing ETL S3 company/tenant breakdown. Raw objects are split by tenant and entity so replay, auditing, and failure isolation stay tenant-safe.

Conceptual raw key shape:

raw/company_id={company_id}/tenant_id={tenant_id}/source_type=kds_cdc/{yyyy}/{MM}/{dd}/batch_id={batch_id}/entity={entity}/...

Raw object format should preserve the Debezium envelope faithfully, such as NDJSON or compressed JSON. The raw object is not just transformed row data; it must contain enough source metadata to replay normalization after a code or mapping fix.

Raw object contents should include:

• original CDC payload,
• operation code,
• source coordinates,
• source table and primary key,
• resolved company_id and tenant_id,
• SQS message id/group/sequence metadata,
• batch id and worker metadata.

SQS messages are not deleted immediately after polling. They are deleted only after:

1. raw S3 write succeeds,
2. durable internal job or batch metadata is created.

After that point, S3 raw plus internal job metadata is the replay source of truth.

Stage 5: Durable Internal Job Creation

After raw S3 succeeds, the worker creates durable internal metadata for downstream ETL work.

This metadata should connect:

• raw batch id,
• company id,
• tenant id,
• tenant database,
• entity/table,
• source type kds_cdc,
• ingestion mode Cdc,
• pipeline stage,
• S3 raw path,
• Debezium coordinate range,
• SQS sequence range,
• status and retry information.

The durable job boundary is important because SQS messages can be deleted only after the system has enough internal state to resume from S3 raw.

Stage 6: Normalize To S3 Load (normalize_load)

normalize_load reads S3 raw and writes load-ready Parquet files to S3 load.

The stage maps CDC envelopes into rows shaped for the existing KDS staging tables:

• staging.kds_orders,
• staging.kds_items,
• staging.kds_item_bumps,
• staging.kds_customers.

Conceptual load key shape:

load/company_id={company_id}/tenant_id={tenant_id}/entity={entity}/batch_id={batch_id}/created_at={timestamp}/...

Normalization behavior:

• split output by tenant and entity,
• preserve source is_deleted,
• preserve CDC source coordinates in metadata columns or batch diagnostics,
• deduplicate exact duplicate CDC messages by source coordinates and primary key,
• keep global order metadata for downstream validation,
• quarantine record-local parse/mapping errors and continue with good records,
• fail the tenant/entity batch only when a safe load artifact cannot be produced.

The load artifact is an optimization checkpoint. S3 raw remains the source of truth.

Stage 7: Load To Source-Shaped Staging (load_to_staging)

load_to_staging resolves the tenant Redshift database and runs Redshift COPY from S3 load Parquet into source-shaped staging tables.

The selected approach does not introduce separate CDC landing tables in the first version. It loads directly into the current KDS staging model.

Load behavior:

• use Redshift COPY, not row-by-row inserts,
• load one tenant/entity artifact at a time,
• apply idempotent delete/insert or merge behavior around the copied batch,
• preserve is_deleted for soft-delete handling,
• validate global ordering before applying the batch,
• fail safely if a gap or out-of-order sequence is detected,
• retry idempotently from the same S3 load artifact if COPY fails.

This stage must never load records whose tenant mapping failed or was ambiguous.

Stage 8: Staging To Curated (staging_to_curated)

staging_to_curated transforms source-shaped KDS staging rows into curated warehouse tables.

Current curated targets include:

• curated.kitchen_orders,
• curated.kitchen_order_items,
• curated.kitchen_item_status_events,
• curated.customers where applicable.

Curated behavior:

• apply idempotent upsert semantics,
• preserve source system identity as production KDS CDC, not synthetic KDS,
• apply soft deletes from source is_deleted,
• keep transaction sizes bounded by batch size,
• do not advance checkpoint if curated merge fails,
• allow retry from staging/load/raw depending on failure point.

Stage 9: Curated To S3 (curated_to_s3)

KDS CDC includes curated-to-S3 export from the start.

This stage exports updated curated Redshift tables to S3 curated Parquet, matching the current ETL architecture and enabling cross-tenant/Athena query patterns.

Conceptual curated key shape:

curated/company_id={company_id}/tenant_id={tenant_id}/entity={entity}/dt={date}/batch_id={batch_id}/...

Export behavior:

• use Redshift UNLOAD,
• export after curated merge succeeds,
• preserve tenant/entity partitioning,
• mark stage success only after export succeeds,
• clean up incomplete curated artifacts on failure when possible,
• retry from curated stage output without re-reading SQS.

Batch Lifecycle And Lineage

Each of the 5 framework stages (extract_raw, normalize_load, load_to_staging, staging_to_curated, curated_to_s3) has its own IngestionBatch lifecycle. The worker does not create one batch for the whole pipeline.

Per-Stage Batch Creation

Every stage calls IIngestionBatchLifecycleService.StartStageBatchAsync before doing work and CompleteStageBatchAsync after success. On failure, StageExecutionSupport.FailStageBatchAsync marks the batch as Failed and writes a failed package.

Each stage batch carries:

• BatchId (new per stage),
• CompanyId, TenantId, TenantDatabase,
• IngestionMode = Cdc,
• SourceType = KdsCdc,
• PipelineStage = the current stage name,
• BusinessDate,
• RawS3Path / LoadS3Path where applicable,
• Status = Pending -> Running -> Succeeded or Failed.

Upstream Batch Chaining

IngestionBatch.UpstreamBatchId links each stage to its predecessor:

extract_raw batch
  └── normalize_load batch (UpstreamBatchId = extract_raw BatchId)
        └── load_to_staging batch (UpstreamBatchId = normalize_load BatchId)
              └── staging_to_curated batch (UpstreamBatchId = load_to_staging BatchId)
                    └── curated_to_s3 batch (UpstreamBatchId = staging_to_curated BatchId)

This lineage enables replay from any stage by following the chain back to the raw batch. It also enables audit queries like "which raw batch produced this curated row."

Explicit Batch Completion

After each stage succeeds, the worker calls CompleteStageBatchAsync to mark the batch as Succeeded. Checkpoint advancement via StageExecutionSupport.AdvanceCheckpointAsync happens after completion, not before. If a stage fails, the batch is marked Failed and subsequent stages do not run.

BusinessDate Determination

CDC has no natural business date — it processes ongoing changes. The adapter uses the processing date:

BusinessDate = DateOnly.FromDateTime(DateTime.UtcNow)

This matches the framework requirement that EtlStageResult returns a DateOnly BusinessDate. The actual CDC progress is tracked in CheckpointValue (Debezium offset/source coordinates), not in BusinessDate. Daily rotation of BusinessDate is acceptable because:

• IngestionBatch.ReceivedAt already captures the exact processing timestamp,
• IngestionCheckpoint.CheckpointValue holds the real Debezium offset for replay,
• checkpoint keys rotate daily per tenant/source/stage, matching the Brink POS pattern.

Checkpointing And Replay

Checkpointing must use source coordinates, not wall-clock time alone.

Checkpoint metadata should include:

• Debezium connector/source identity,
• source database/schema/table,
• Debezium offset/source coordinates,
• SQS message group id,
• SQS sequence number or message id,
• global order range,
• last successful raw batch id,
• last successful load batch id,
• last successful curated batch id,
• last successful curated export batch id.

S3 raw is the replay source of truth. Replay must support:

Checkpoints advance only after the configured durable success point for the stage. If SQS messages have already been deleted, replay must never depend on SQS retention.

Quarantine And Failed Packages

Bad records are quarantined; good records continue when safe.

Quarantine applies to record-local failures such as:

• malformed Debezium envelope,
• missing source primary key,
• missing or ambiguous tenant mapping,
• unsupported source table,
• invalid value that prevents row mapping,
• duplicate message that cannot be reconciled safely.

Batch-level failure applies when safe partial processing is not possible, such as:

• raw S3 write failure,
• internal job metadata creation failure,
• ordering gap or out-of-order condition,
• Parquet artifact cannot be produced safely,
• Redshift COPY fails,
• curated merge fails,
• curated export fails.

Quarantine/failed package metadata should include:

• failure reason,
• source payload or pointer to raw payload,
• source table and primary key,
• Debezium operation and coordinates,
• SQS message id/group/sequence,
• resolved or unresolved company/tenant mapping,
• raw S3 path,
• load S3 path when available,
• pipeline stage,
• batch id,
• retry/replay guidance.

Supported Corner Cases

Missing Tenant Mapping

If source customer/account or store_guid cannot be mapped, the record is quarantined. It is not written to any tenant Redshift database. Other valid records continue.

Ambiguous Tenant Mapping

If one source identity maps to multiple possible tenants, the record is quarantined. No best-effort tenant selection is allowed.

Bad Record In Good Batch

A malformed or unmapped record is quarantined when the failure is record-local. Good records in the same poll window continue through raw, load, staging, curated, and export.

Raw S3 Write Failure

If raw S3 write fails, SQS messages are not deleted. SQS visibility timeout and retry handle redelivery.

Internal Job Creation Failure

If raw S3 succeeds but durable internal job creation fails, SQS messages are not deleted. A retry may write duplicate raw objects; downstream dedupe must use source coordinates and batch metadata.

Duplicate CDC Message

Duplicate messages are detected by Debezium source coordinates plus source table and primary key. Replays must be idempotent.

Out-Of-Order Or Missing Sequence

Global order is required. If the worker detects a sequence gap or out-of-order batch, it must fail safely before Redshift load and preserve diagnostics for replay.

Source Delete Or Soft Delete

Source is_deleted is preserved. Curated rows are soft-deleted rather than physically deleted in the normal path.

Redshift COPY Failure

Raw and load artifacts remain durable. Retry uses the same S3 load artifact and must be idempotent.

Curated Merge Failure

Checkpoint does not advance. The system retries after SQL/data correction from staging/load/raw depending on the failure.

Curated Export Failure

Curated Redshift data remains the committed warehouse state. The export stage can retry UNLOAD; incomplete S3 curated artifacts should be cleaned up when possible.

SQS Message Redelivery After Raw Write

If SQS redelivers messages after raw write, dedupe uses source coordinates and SQS metadata. Duplicate raw batches must not corrupt staging or curated state.

Low Volume Tenant

The batch age limit closes a batch even when record count is below target, so quiet tenants still receive data within the configured latency window.

High Volume Tenant

The record count and byte limits bound transaction size and worker memory. Large streams become multiple ordered micro-batches.

Ongoing CDC Without Snapshot

This feature only handles ongoing CDC. Production use assumes a baseline seed or backfill exists before CDC is considered complete history.

Non-Goals

• Do not reuse the simulator publisher as the production CDC ingestion path.
• Do not identify production CDC as SyntheticKds.
• Do not create one ingestion batch per CDC record.
• Do not rely on SQS as long-term replay after messages are acknowledged.
• Do not introduce CDC landing tables in the first version.
• Do not physically delete curated rows in the normal KDS CDC path.