# Cheat sheet — CDC / Streaming-ETL (this repo)

Dense revision sheet. One glance per lesson. Interview one-liners at the bottom.

---

## The pipeline (say it out loud)

```
service Postgres  ──WAL──▶  Debezium source connector  ──Avro──▶  Kafka topic
  (pgoutput,               (io.debezium…PostgresConnector)      local.manabie.bob.public.users
   slot, publication)                                                    │
                                                    ┌───────────────────┴───────────────────┐
                                     (plain sync)   │                          (warehouse)   │
                                          JDBC sink connector          ksqlDB stream/table/join
                                     unwrap + route → upsert(pk)         → CREATE SINK CONNECTOR
                                                    │                                         │
                                            another service's Postgres            dim_*/fact_* warehouse
                                             (bob → auth.users)                    (→ chained CDC → partner DWH)
```
Reconciled by **hephaestus** (git-as-desired-state). **Three stacks:** data-sync · KEC DWH · lmsdwh.

## Debezium source (the front door)

```
connector.class = io.debezium.connector.postgresql.PostgresConnector
plugin.name     = pgoutput                 # native logical decoding (needs wal_level=logical)
slot.name       = {env}_{org}_{db}         # replication slot retains WAL; slot.drop.on.stop=false
publication.name= debezium_publication     # publication.autocreate.mode = disabled  → made by SQL migration
snapshot.mode   = never | initial          # existing rows; "never" ⇒ rely on incremental snapshots
signal.data.collection = public.dbz_signals            # signal table
incremental.snapshot.chunk.size = 512
tombstones.on.delete = true                # null message after a delete
transforms.route → RegexRouter  bob.public.users → local.manabie.bob.public.users
```
Source golib: `internal/platform/datapipeline/source.go:145-198` · live `…/kafka-connect-source-connector/bob.json`.

## Snapshots

- **initial** — read the whole table once on first connect, then stream the WAL.
- **incremental** — snapshot existing rows *while streaming*, in resumable chunks, triggered by a
  **signal** → the way to add a table to CDC **without recreating the connector**.
- Trigger (`debezium.IncrementalSnapshot`, `incremental_snapshot.go:39-72`): wait for the slot to be
  active, then `INSERT INTO public.dbz_signals VALUES (id, 'execute-snapshot', '{"data-collections":[…]}')`.

## JDBC sink (the destination)

```
connector.class = io.confluent.connect.jdbc.JdbcSinkConnector
insert.mode = upsert   pk.mode = record_value   pk.fields = user_id   auto.create = false
transforms = unwrap, route
  unwrap = io.debezium.transforms.ExtractNewRecordState    # flatten envelope → the "after" row
  route  = RegexRouter  local.manabie.bob.public.users → $5 = "users"
delete.enabled = false ; unwrap.delete.handling.mode = drop   # deletes DROPPED (soft-delete via deleted_at)
# fields.whitelist = INTERSECTION of source & sink columns ; no DLQ configured
```
Sink golib: `internal/platform/datapipeline/sink.go` · live `…/kafka-connect-gen-sink-connector/bob_to_auth_users.json`.

## Generation + hephaestus (the repo signature)

```
generated_connectors/bob.yaml   (source_db + envs + orgs + pipelines[{source_table, sinks:[{database}]}])
        │  gen-data-pipeline  (datapipeline.go GenerateSources/GenerateSinks)
        ▼  fan out: env × org × table × sink  →  hundreds of connector JSON files (bob = 169/237 sinks)
hephaestus_upsert_kafka_connect  (a gjob, run as a Helm hook)
   reconcile:  live cluster (REST :8083)  vs  JSON files (git = desired state)
   Upsert:  GET → create if absent, else PUT only if config changed (diff)
   Delete-guard: only io.confluent…JdbcSinkConnector ; NEVER sources (slot loss) or ksqlDB (all-caps)
   if --sendIncrementalSnapshot & source & new table → fire dbz_signals snapshot (no recreation)
```
Connect client: `internal/golibs/kafkaconnect/connector.go` (`Upsert` :84, `TryUpdateConnector` :238,
`Delete` :137, `IsConnectorReady` :379). Deploy hooks 2-init-ksql → 3-upsert → 4-validate → 5-migrate-ksql → 6-sync-dwh.

## ksqlDB (transform)

- **stream** = event log (append-only) · **table** = latest value per key (materialized view).
- Transforms in versioned SQL (`datawarehouses/lms/ksql/migrations/V*.sql`): `CREATE STREAM/TABLE …
  VALUE_FORMAT='AVRO'`, `LATEST_BY_OFFSET(...)`, JOINs (dim_user = users ⋈ user_group_member ⋈ user_group),
  then `CREATE SINK CONNECTOR`. ksqlDB owns those sinks (all-caps → hephaestus won't delete).
- "ksqlDB migrations" = Confluent `ksql-migrations` CLI; `PrepareMigration` (`init_migrate.go`) seeds
  its `MIGRATION_EVENTS` bookkeeping stream.

## Warehouse (lmsdwh)

- **Star schema:** `dim_*` (dimensions) + `fact_*` (facts) — `internal/lmsdwh/entities/`, `migrations/lmsdwh/`.
- **Chained CDC:** lmsdwh is itself a CDC source → partner DWHs; multi-tenant `Filter$Value` JSONPath
  on `resource_path` at the partner sink. Accuracy verifiers reconcile row counts.

## Ops

- **Add a table:** add `public.<t>` to source `table.include.list` + a `pipelines:` entry in `<db>.yaml`
  → regen → `hephaestus_upsert_kafka_connect --sendIncrementalSnapshot` (back-fills, no recreation).
- **Health:** JMX→Prometheus; **auto-heal cron** every 10 min restarts FAILED/UNASSIGNED tasks; Slack
  alert on connector deletion; `IsConnectorReady` = connector RUNNING + all tasks RUNNING.
- **Local:** `./local/run.bash start` regenerates connector dirs + brings up the Connect service; Kafka UI :8082.

---

## Interview one-liners

- **"What is CDC?"** — turning a DB's committed changes into an event stream. We do log-based CDC:
  Debezium reads the Postgres WAL via `pgoutput`, so we catch every insert/update/delete with ~no load.
- **"Source vs sink connector?"** — a source (Debezium) reads the WAL → Kafka; a sink (JDBC) reads a
  topic → a database. Both run on Kafka Connect.
- **"Why a replication slot / publication?"** — the slot retains WAL so we never miss changes across
  restarts; the publication declares which tables — and we manage it via SQL migration, not Debezium.
- **"Initial vs incremental snapshot?"** — initial loads the whole table on connect; incremental
  snapshots existing rows in chunks *while streaming*, on a signal — how we add a table without
  recreating the connector (and losing the slot).
- **"How is the sink idempotent?"** — JDBC upsert on the primary key. Delivery is at-least-once, so an
  idempotent upsert makes reprocessing safe.
- **"Why generate 248 connectors?"** — you can't hand-maintain them. We declare pipelines in YAML,
  generate the JSON (env×org×table×sink), and hephaestus reconciles the cluster to match — git as
  desired state, with a guard that never deletes a source (slot loss) or a ksqlDB connector.
- **"Dual-write problem?"** — writing to the DB and publishing to Kafka separately can half-fail. CDC
  derives the event from the committed WAL, so the DB is the single source of truth.