KDS CDC Branch Direction And Process Flow

Purpose

This document records the assumptions, process flow, design decisions, and implementation direction taken on this branch for KDS CDC ingestion and ETL.

It is intended as historical context for later analysis and maintenance. It describes what this branch is trying to achieve, how the current implementation approaches it, which assumptions it relies on, and which decisions were made while moving from simulator-shaped CDC intake to Debezium-native intake.

What We Wanted To Achieve

The branch goal is to ingest Logic Control KDS change data capture events into the KDSAI ETL and warehouse pipeline using real Debezium payloads.

Primary outcomes:

• Treat Debezium payloads as the only supported KDS CDC intake contract.
• Stop depending on simulator-specific message shape, mock contract versioning, and producer-generated ordering semantics.
• Keep the existing KDSAI ETL stage model instead of building a completely separate ingestion system.
• Preserve source payload and source coordinates in raw storage so replay and audit remain possible.
• Normalize KDS CDC rows into source-shaped staging tables for orders, items, item bumps, and customers.
• Build curated warehouse state from staging state, not from a fragile assumption that CDC events arrive parent-before-child.
• Make SQS a transport buffer, not the long-term source of truth.

The desired long-term shape is capture-first and interpret-later:

SQS Debezium message
  -> durable raw capture
  -> tenant resolution and promotion
  -> normalize_load
  -> load_to_staging
  -> staging_to_curated
  -> curated_to_s3

Branch-Level Direction

This branch changes KDS CDC from a mock/simulator contract toward a production Debezium contract.

Important direction decisions:

• Debezium envelope shape is authoritative.
• Top-level Debezium fields schema and payload are expected.
• Semantic data comes from payload.source, payload.op, payload.after, and payload.before.
• Row data is not read from the envelope root.
• payload.after is used for create, update, and snapshot/read events.
• payload.before is used for delete events.
• Primary key field is guid for supported KDS tables.
• Supported tables are orders, items, item_bumps, and customers.
• Unsupported tables are rejected before tenant-scoped ETL work.
• SQS FIFO metadata can be preserved but is not required for correctness.
• MessageGroupId is not a partitioning or ordering dependency for KDS CDC.
• SequenceNumber is nullable and optional.
• Debezium source coordinates are the source identity and ordering basis.
• store_guid is the tenant identity for this integration.
• source_account_id is no longer part of required routing identity.

Current Branch Implementation Summary

The branch implements Debezium-native parsing and removes the simulator/mock CDC producer path from the KDS CDC intake surface.

Implemented now:

• Parser accepts Debezium envelopes and rejects the old mock/custom contract path.
• Parser accepts SQS messages with missing MessageGroupId and missing SequenceNumber.
• Parser extracts source database, schema, table, operation, selected row image, primary key, store guid, and source coordinates.
• Duplicate detection keys on Debezium source identity instead of SQS FIFO identity.
• Ordering token parsing understands Debezium coordinates in the form binlog-file:position:row.
• Batch building no longer splits or closes batches on SQS MessageGroupId boundaries.
• Tenant resolution maps by store_guid and no longer requires source_account_id.
• Child messages without direct store_guid become retryable tenant-unresolved messages instead of permanent quarantine.
• Raw records preserve nullable FIFO metadata.
• Normalization reads the selected Debezium row image from payload.after or payload.before.
• Mock CDC producer, mock envelope factory, mock queue publisher, and simulator CDC publishing registration were removed.
• Tests were converted toward real Debezium payload fixtures.

Known implementation boundary:

• The desired separate landing RAW plus explicit tenant RAW promotion state is documented as target architecture, but this branch does not fully introduce a new durable landing RAW store and promotion-state table.
• Current branch still primarily follows the existing tenant-scoped ETL raw path once tenant resolution succeeds.
• Missing child tenant relation state is handled safely by retrying unresolved messages, but a durable order_guid -> store_guid and item_guid -> store_guid relation index is not yet fully introduced as a new persistence model.

Assumptions

Source Payload Assumptions

KDS CDC messages are Debezium messages from Logic Control KDS source tables.

Assumed Debezium shape:

{
  "schema": {},
  "payload": {
    "before": {},
    "after": {},
    "source": {
      "db": "...",
      "schema": "...",
      "table": "...",
      "file": "mysql-bin-changelog.237363",
      "pos": 123,
      "row": 0
    },
    "op": "c"
  }
}

Supported operation codes:

• c means create and uses payload.after.
• u means update and uses payload.after.
• r means snapshot/read and uses payload.after.
• d means delete and uses payload.before.

Supported table names:

• orders
• items
• item_bumps
• customers

Assumed row identity:

• each supported row image has a primary key field named guid.
• orders and customers normally carry direct store_guid.
• items may not carry direct store_guid and may need resolution through order_guid.
• item bumps may not carry direct store_guid and may need resolution through item_guid.

Tenant Assumptions

For this KDS CDC integration, store_guid is treated as the tenant identifier.

Routing assumptions:

• store_guid == tenant_id for KDS CDC routing purposes.
• store_guid is stable and unique enough to resolve tenant ownership.
• A mapping row still connects source type and store guid to platform company/tenant warehouse metadata.
• source_account_id is optional historical metadata and not required for KDS CDC tenant resolution.

Child resolution target model:

• direct store_guid should be used when present.
• items without store_guid should resolve through order_guid -> store_guid.
• item bumps without store_guid should resolve through item_guid -> store_guid.
• unresolved child records should not be treated as bad data immediately.
• unresolved child records should be retried after parent relation state is available.

Transport Assumptions

SQS remains the transport mechanism. It may be a FIFO queue, but application correctness must not rely on FIFO-only attributes.

Transport assumptions:

• SQS MessageId and ReceiptHandle are required to process and acknowledge a transport message.
• SQS MessageGroupId is optional for KDS CDC logic.
• SQS SequenceNumber is optional for KDS CDC logic.
• Approximate receive count is useful diagnostics but not business identity.
• Source identity must come from Debezium source metadata, table, primary key, and source coordinates.

Ordering Assumptions

Ordering is best represented by Debezium source coordinates, not SQS FIFO sequence.

Ordering token assumptions:

• coordinate format is file:pos:row.
• binlog file ordinal is parsed from the numeric suffix of the file name.
• position and row are numeric components.
• a comparable token can be built from file ordinal, position, and row.
• missing or unparsable coordinates should not block raw capture.
• comparable decreasing coordinates indicate unsafe ordering for a stream.

This branch treats missing comparable source coordinates as acceptable for intake so durable capture is not blocked. Later ETL stages can still use preserved metadata to decide whether a batch is safe to apply.

Process Flow

1. SQS Receive

The KDS CDC worker receives transport messages from SQS.

Current behavior:

• request SQS message body and attributes when present.
• preserve MessageGroupId and SequenceNumber when present.
• allow MessageGroupId and SequenceNumber to be null.
• keep body length for payload-size validation.

The worker does not treat one SQS message as one full ETL batch. It builds bounded micro-batches using existing batching rules such as count, age, and buffered byte limits.

2. Parse Debezium Envelope

The parser validates Debezium structure and extracts KDS CDC metadata.

Validation outcomes:

• malformed JSON is rejected.
• missing Debezium payload is rejected.
• missing payload.source.db or payload.source.table is rejected.
• unsupported table is rejected.
• unsupported op code is rejected.
• missing selected row image is rejected.
• missing guid primary key is rejected.

Accepted message fields:

• source database from payload.source.db.
• source schema from payload.source.schema when present.
• source table from payload.source.table.
• operation from payload.op.
• selected row image from after or before.
• primary key name guid and primary key value from selected row image.
• direct store_guid when present and non-blank.
• source coordinates from file, pos, and row when all are present.
• body hash for duplicate diagnostics.

3. Duplicate Detection

Duplicates are detected using source-level Debezium identity rather than FIFO metadata.

Branch dedupe identity:

source database + source table + primary key value + source coordinates

Fingerprint data includes operation, SQS message id, and body hash. FIFO group and sequence are not part of correctness identity.

Decision:

• same Debezium identity with same fingerprint can be safely treated as duplicate delivery.
• same Debezium identity with different fingerprint is suspicious and should not be silently applied.

4. Ordering Validation

Ordering validation uses structured Debezium coordinates when available.

Coordinate comparison model:

mysql-bin-changelog.237363:1615:0
  -> file ordinal 237363
  -> position 1615
  -> row 0

Comparable token shape:

fileOrdinal:position:row

Decision:

• decreasing comparable tokens are unsafe and fail the batch path.
• duplicate comparable tokens are quarantined as suspicious.
• missing or unparsable comparable tokens are accepted at intake because capture should not be blocked.

5. Micro-Batch Construction

The batch builder groups accepted messages into micro-batches.

Branch decision:

• batches are not partitioned by SQS MessageGroupId.
• a changed MessageGroupId does not force batch close.
• source accounts are optional and omitted when empty.
• stores and entities are collected from accepted messages.

This keeps batching aligned with Debezium semantics and avoids coupling the ETL pipeline to transport-level FIFO grouping.

6. Tenant Resolution

Tenant resolution happens before tenant-scoped raw storage and warehouse loading.

Branch behavior:

• direct store_guid is used as routing identity.
• mapping repository lookup uses source type and store guid.
• source_account_id is no longer required.
• missing mapping is still a failure for direct-tenant messages.
• ambiguous mapping is still a failure.
• missing or inactive tenant warehouse is still a failure.
• missing direct store_guid returns TenantUnresolved instead of permanent quarantine.

Reason for this decision:

• orders and customers can usually resolve directly.
• child records can arrive before parent relation state is available.
• unresolved children should not be destroyed or quarantined as bad data if they may become resolvable later.

Current limitation:

• durable relation indexes for order_guid -> store_guid and item_guid -> store_guid are target architecture but not fully implemented in this branch.
• unresolved child messages currently remain retryable rather than being promoted through a completed landing RAW relation-resolution workflow.

7. Raw Capture And Handoff

Target architecture says raw capture should happen as early as possible in a landing RAW area, before tenant resolution is fully reliable.

Target flow:

valid Debezium message
  -> landing RAW object
  -> promotion state
  -> SQS ack
  -> tenant resolution and tenant RAW promotion

Current branch behavior is a smaller step toward that target:

• raw writing remains integrated with the existing extract_raw stage after tenant resolution.
• raw records now preserve Debezium envelope, selected metadata, source coordinates, nullable FIFO transport metadata, and original payload.
• SQS ack behavior still depends on successful handoff for resolved slices.
• unresolved messages are not acknowledged and get visibility extension.

Decision record:

• durable raw-first capture is the desired architecture.
• branch implementation makes the code Debezium-safe and transport-decoupled first.
• full landing RAW plus promotion-state persistence remains a follow-up if strict capture-before-resolution is required.

8. Normalize Load

normalize_load reads raw NDJSON records and writes normalized load artifacts.

Important branch change:

• normalization maps from selected Debezium row image, not from the envelope root.

Selected image rule:

op == d  -> payload.before
else     -> payload.after

Normalization output entities:

• orders to KDS order staging row shape.
• items to KDS item staging row shape.
• item_bumps to KDS item bump staging row shape.
• customers to KDS customer staging row shape.

Normalization preserves:

• primary key.
• operation.
• delete marker.
• source database, table, and coordinates.
• nullable SQS metadata.
• raw envelope for replay and audit.

Unknown source columns are ignored by row mapping but remain preserved in raw payload.

9. Load To Staging

load_to_staging loads normalized artifacts into tenant Redshift staging tables.

The branch continues using existing staging-oriented ETL behavior.

Intended role of staging:

• hold source-shaped KDS state.
• provide tenant-local relational state for downstream curated rebuild.
• support idempotent reprocessing from durable raw or load artifacts.

10. Staging To Curated

Curated state should be rebuilt from staging truth, not from the order in which SQS delivered messages.

Decision:

• CDC arrival order should not be the only mechanism that makes parent and child rows consistent.
• if an item arrives before an order, curated correctness should come from staged tenant state after both are present.
• curated jobs should operate from source-shaped staging tables and rebuild affected targets.

Curated targets covered by this branch family include:

• kitchen orders.
• kitchen order items.
• kitchen item status events.
• customers.

11. Curated To S3

curated_to_s3 remains part of the KDS CDC ETL flow.

Purpose:

• export curated tenant state to S3.
• preserve the same externalized analytics pattern used by other ETL adapters.
• keep Redshift curated state and S3 curated artifacts aligned through stage metadata.

Quarantine And Retry Decisions

The branch separates bad data from temporarily unresolved routing.

Rejected or quarantined:

• malformed JSON.
• missing Debezium payload.
• missing required Debezium source metadata.
• unsupported table.
• unsupported operation.
• missing selected row image.
• missing primary key.
• ambiguous tenant mapping.
• missing tenant warehouse.
• inactive tenant warehouse.
• suspicious duplicate.
• unsafe decreasing ordering token.

Retryable unresolved:

• child message without direct store_guid where tenant may later resolve through relation state.

This distinction is important because unresolved child routing is expected in CDC streams and should not be treated the same as invalid payload format.

Removed Simulator And Mock CDC Path

The branch removes the KDS CDC mock producer path because it conflicted with the Debezium-only direction.

Removed concepts:

• mock contract version branch in parser.
• mock envelope factory.
• mock CDC producer.
• mock sequence allocator.
• mock SQS queue publisher.
• simulator publishing KDS CDC mock messages.
• tests that validate mock message generation rather than Debezium intake.

Decision:

• simulator may still create synthetic warehouse rows for demo traffic.
• simulator no longer defines or publishes the KDS CDC intake contract.
• KDS CDC tests should use real Debezium-style fixture payloads.

Data And Persistence Decisions

Source tenant mapping changed from source-account keyed routing to store-guid keyed routing.

Mapping direction:

source_type + store_guid -> company_id + tenant_id + tenant warehouse

Entity changes:

• KdsCdcSourceTenantMapping.SourceAccountId is nullable.
• repository lookup no longer requires source account.
• active mapping index is based on source type, store guid, and active state.
• quarantine transport fields allow nullable message group and sequence.

Reason:

• Debezium payload samples do not establish source_account_id as required routing identity.
• KDS CDC tenant identity is the store guid.
• SQS FIFO metadata should not be forced into durable data as required business metadata.

What This Branch Does Not Claim To Solve Completely

This branch is a major contract and direction change, not a complete redesign of every ETL persistence boundary.

Not fully complete in this branch:

• separate immutable landing RAW area before tenant resolution.
• explicit durable promotion-state entity for landing RAW to tenant RAW.
• persistent relation index tables for child tenant resolution.
• full ack-after-landing behavior independent of tenant resolution.
• replacement synthetic data generator that emits real Debezium envelopes.
• admin UI or operations tooling for unresolved landing records.

These are still valid future work if the system needs the fully documented capture-first architecture.

Why These Decisions Were Made

The implementation favored the fastest safe movement toward production compatibility.

Reasons:

• real Debezium samples showed the previous simulator contract was not representative.
• relying on MessageGroupId and SequenceNumber would make transport metadata part of business correctness.
• source coordinates are better replay, dedupe, audit, and ordering identity than SQS attributes.
• store_guid is the stable KDS tenant identity and avoids carrying a synthetic source-account requirement.
• normalizing from the wrong JSON level would silently produce bad data once real Debezium messages arrive.
• removing mock producer code prevents old simulator behavior from continuing to look like a supported intake contract.

Maintenance Notes

Future maintainers should treat this branch as the point where KDS CDC changed from simulator-compatible to Debezium-compatible.

When extending this area:

• preserve Debezium payload as the supported contract.
• avoid adding new logic that requires SQS MessageGroupId or SequenceNumber for correctness.
• keep source coordinates in raw, diagnostics, checkpoints, and replay identity.
• route by store_guid and relation indexes, not by source account.
• make child tenant-unresolved behavior retryable unless the payload itself is invalid.
• keep normalization tied to payload.after or payload.before based on operation.
• prefer durable capture and replayable state over doing all semantic work inside the SQS polling loop.

Summary

This branch establishes the Debezium-native direction for KDS CDC.

The biggest conceptual shift is that KDS CDC intake is no longer a simulator-shaped stream with FIFO-group ordering and source-account tenant routing. It is a Debezium stream where source metadata, row images, source coordinates, and store_guid drive correctness.

The branch implements the core compatibility changes needed for real payloads and removes mock CDC generation. It also records the next architectural boundary: durable landing RAW plus explicit tenant RAW promotion, which remains the intended long-term shape for safest capture-first operation.