# Repo map — CDC / Streaming-ETL (ground truth)

The survey behind this course. Every lesson's `file:line` anchors resolve here. When code moves, fix
this file first, then the lessons. Grouped by the flow: source → Kafka → sink → warehouse, plus the
hephaestus control plane.

**The one framing to hold first:** there are **THREE distinct CDC stacks** (three Connect clusters,
three schema registries, three ksqlDB servers) — do NOT teach it as one cluster:
1. **Cross-service data-sync** (`platforms` stack, the biggest) — Debezium captures each service's
   Postgres; ~237 JDBC sink connectors replicate specific tables INTO *other services'* Postgres
   (bob→auth, bob→eureka…). Operational data federation, **no warehouse**.
2. **KEC data warehouse** (`data-warehouse` stack) — CDC → **ksqlDB** transforms (joins/aggregations)
   → JDBC sink into a warehouse.
3. **lmsdwh** (LMS analytics warehouse) — dedicated star-schema DWH; **chained CDC**: services →
   lmsdwh Postgres → (CDC again) → partner DWHs.
hephaestus reconciles all three.

---

## 1. Kafka Connect — the framework/runtime

- **Deployment: distributed mode, a customized Debezium image.** Local: `local/docker-compose.infra.yaml:193-220`
  — image `customized_debezium_connect:2.4.1.Final.…` (NOT vanilla Confluent), distributed via
  `CONFIG/OFFSET/STATUS_STORAGE_TOPIC` (`:204-206`), REST API `:8083`. Secrets via
  `FileConfigProvider` (`:213-214`) → `${file:/config/…:password}`. Kafka UI `:8082` (`:180-191`).
  Prod: `deployments/helm/platforms/kafka-connect/templates/deployment.yaml` (+ `data-warehouse/` and
  `lmsdwh/` clusters). Schema Registry: Confluent `cp-schema-registry:8081`, Avro, compatibility
  `none` (`docker-compose.infra.yaml:162-178`).
- **The Connect client golib** `internal/golibs/kafkaconnect/connector.go` (wraps `go-kafka/connect`):
  `ConnectClient` (`:23-31`, Get/Create/Update/Delete/List/status), `ConnectorManagement` (`:37-41`,
  `Upsert`/`Delete`/`GetListConnectorNames`).
  - `Upsert` (`:84-125`): per connector, GET → absent? `TryCreateConnector` (`:209`) : `TryUpdateConnector`
    (`:238`); `try.Do` retry.
  - `TryUpdateConnector` (`:238-300`): PUT only if `anyChangeInConfig` (`util.go:191-200`); wait
    `IsConnectorReady` (`:379-397`, connector RUNNING + all tasks RUNNING); if `SendIncrementalSnapshot`
    and it's a source, compute newly-captured tables (`GetNewCapturedCapture` `:335-352`) → fire an
    incremental-snapshot signal (`:264-297`).
  - `Delete` (`:137-195`): **only deletes `io.confluent.connect.jdbc.JdbcSinkConnector`** — never
    sources (deleting drops the publication → data loss, comment `:154-156`) or all-caps
    ksqlDB-managed connectors (`:159`).
  - `util.go`: `extractTableFromTopic` regex `([^.]+).([^.]+).([^.]+).([^.]+).([^.]+)` maps
    `stag.manabie.bob.public.users` → {db,schema,table} (`:98-114`); `ignorableField` whitelist
    (`:159-170`) = fields that don't trigger a connector update.
- Converters: **Avro on all CDC topics** via Confluent `AvroConverter` + registry
  (`source.go:160-165`). `linkedin/goavro/v2` is NOT in the connect path (only a separate eureka
  consumer).

## 2. Debezium — the CDC source

- **`internal/golibs/debezium/`**
  - `incremental_snapshot.go` — `IncrementalSnapshot(ctx, db, signalTable, data)` (`:39-72`): sleep 10s,
    poll `pg_replication_slots.active` (`isReplicationSlotActive` `:74-82`), then
    `INSERT INTO <signalTable> VALUES ($1,$2,$3)` = (id=sourceID+ULID, `'execute-snapshot'`, data).
    `DataCollection.String()` (`:27-33`) → `{"data-collections": ["public.tc",…]}`. Comment `:16-19`:
    "snapshot new added tables **without recreating** the source connector."
  - `subscription.go` — the NATS-JetStream-driven design is **dead code** (handler commented out,
    `:108-142`); today the signal is a **direct SQL INSERT** from `TryUpdateConnector`.
- **Source config** (canonical: `internal/platform/datapipeline/source.go:145-198` `NewSourceJSON`;
  live: `local/configs/services/kafka-connect-source-connector/bob.json`):
  - `connector.class = io.debezium.connector.postgresql.PostgresConnector` (`:150`);
    `plugin.name = pgoutput` (`:158`, the native logical-decoding plugin).
  - `slot.name = {env}_{org}_{db}` e.g. `local_manabie_bob`; `slot.drop.on.stop = false` (`:166-167`).
  - `publication.name = debezium_publication`; **`publication.autocreate.mode = disabled`** (`:168-169`)
    — publications are created by **SQL migrations**, not Debezium (`migrations/lmsdwh/1001_migrate.up.sql:15-53`
    `CREATE PUBLICATION` / `is_table_in_publication()` / `ALTER PUBLICATION … ADD TABLE`).
  - `snapshot.mode = never` (generated default `:170`; bob.json overrides to `initial`).
  - `tombstones.on.delete = true` (`:171`); heartbeat every 20s writing `debezium_heartbeat`
    (`:172-173`) to keep the slot advancing.
  - `signal.data.collection = public.dbz_signals` (`:177`); `incremental.snapshot.chunk.size = 512`
    (`:180`); source `route` RegexRouter `([^.]+).([^.]+).([^.]+)` → `local.manabie.$1.$2.$3` (`:187-190`).
- **Captured DBs:** bob, calendar, eureka, fatima, invoicemgmt, lessonmgmt, mastermgmt, entryexitmgmt.
  **bob is the hub** — 169 of 237 sinks read bob (`bob.json:30` ~90 tables).

## 3. The sink side

- **JDBC sink** (`internal/platform/datapipeline/sink.go`; live:
  `local/configs/services/kafka-connect-gen-sink-connector/bob_to_auth_users.json`):
  - `connector.class = io.confluent.connect.jdbc.JdbcSinkConnector` (`sink.go:171`); `connection.url`
    = `${file:…:auth_url}` (another service's Postgres).
  - `insert.mode = upsert`, `pk.mode = record_value`, `pk.fields = user_id`, `auto.create = false`.
  - SMT chain `transforms = unwrap,route` (`:181`): `unwrap = io.debezium.transforms.ExtractNewRecordState`
    (flattens the Debezium envelope, `:185`); `route = RegexRouter … → $5` (strips
    `local.manabie.bob.public.users` → `users`, `:182-184`).
  - **Deletes dropped:** `delete.enabled = false`, `unwrap.drop.tombstones = true`,
    `unwrap.delete.handling.mode = drop` (`:186,191-192`) — soft-delete via `deleted_at` instead.
  - **No DLQ** in the generated sinks (a gap worth flagging).
- **Schema-driven generation** (`sink.go`): loads source AND sink table schemas (`initTableSchemas`
  `:65-83`); `pk.fields` must match both (`GetPKFields` `:85-95`); `fields.whitelist` = **intersection**
  of source & sink columns (`GetFieldsWhitelist` `:97-104`). Avoids schema drift. `PreWriteCheck`
  (`:214-254`) guards the literal→Helm-template naming migration.

## 4. Generation + hephaestus (the control plane) — the repo signature

- **The `datapipeline` abstraction** `internal/platform/datapipeline/datapipeline.go`: declarative
  `Def` = source DB + envs + orgs + `pipelines: [{source_table, sinks:[{database}]}]` (`:12-30`).
  `NewPipeline` loads YAML (`:32-43`); `GenerateSources`/`GenerateSinks` iterate **env × org × table
  × sink** (`:53-136`) → one connector JSON each. Input defs:
  `deployments/helm/backend/hephaestus/generated_connectors/bob.yaml`. Generator command
  `gen-data-pipeline` (`cmd/utils/data_pipeline_parser/command2.go:13-42`). One `bob.yaml` fans out to
  hundreds of files. **Source-database-centric fan-out.**
- **hephaestus** (`cmd/server/hephaestus/`, `internal/hephaestus/`) — a **job runner** (not a
  connector gRPC server; only a dummy healthcheck). Jobs via `bootstrap.RegisterJob`, run as Helm
  hooks:
  - `hephaestus_upsert_kafka_connect` (`hephaestus.go:6-8` → `upsert_kafka_connect.go:43-157`) — the
    core reconcile: delete stale (Slack-notified `:78-101`) + upsert source/gen-sink/custom-sink dirs
    (`:121-154`). Flags `--deployCustomSinkConnector` / `--sendIncrementalSnapshot`.
  - `hephaestus_validate_kafka_connectors` (`validate_kafka_connectors.go:14-31`) — dry-run diff.
  - `_upsert_lmsdwh_kafka_connector`, `_dwh_accuracy` (`data_warehouse_accuracy.go:62-84`, row-count
    reconcile), `_prepare_ksqldb_migration`, `_migrate_datalake/_datawarehouse/_lmsdwh`.
  - **State model = git-as-desired-state:** no DB of connectors — read JSON from config dirs, diff vs
    the live REST API, converge. `DeleteConnector.DeleteConnectors` (`delete_kafka_connector.go:139-168`)
    = `live − onDisk`.
  - **Deploy = numbered Helm hooks** (`deployments/helm/backend/hephaestus/templates/`): `2_init_ksql` →
    `3_upsert_kafka_connect` (`gjob hephaestus_upsert_kafka_connect --deployCustomSinkConnector
    --sendIncrementalSnapshot`, `:88-92`) → `4_validate` → `5_migrate_ksql` → `6_sync_datawarehouse`.
  - **BDD** (`features/hephaestus/`): `job_upsert_kafka_connector.feature` (bob→fatima sync);
    `incremental_snapshot_new_table.feature:25-32` (add table, back-fill without recreation);
    `recreate_source_connector_not_consume_db_log.feature` (recreate ≠ replay downtime data).

## 5. ksqlDB — stream processing

- `internal/platform/ksqldb/init_migrate.go` (`thmeitz/ksqldb-go`): `PrepareMigration` (`:19-44`)
  bootstraps the **Confluent ksql-migrations** bookkeeping — a `MIGRATION_EVENTS` stream (`:46-61`) +
  a `…_SCHEMA_VERSIONS` table via `LATEST_BY_OFFSET … GROUP BY VERSION_KEY` (`:63-83`). Streams per
  cluster: `MIGRATION_EVENTS`/`KEC_…`/`LMSDWH_…`.
- Actual transforms are versioned SQL:
  `deployments/helm/backend/hephaestus/datawarehouses/lms/ksql/migrations/V*.sql`, e.g.
  `V000002__dim_user.sql`: `CREATE STREAM … WITH (VALUE_FORMAT='AVRO')` from CDC topics (`:7-9`),
  `LATEST_BY_OFFSET(AFTER->…)` aggregation (`:14-47`), **JOIN** users ⋈ user_group_member ⋈ user_group
  (`:134-198`), then `CREATE SINK CONNECTOR … JdbcSinkConnector … table.name.format=public.dim_user`
  (`:200-216`). ksqlDB owns those sink connectors (all-caps → Go `Delete` won't touch, `connector.go:159`).
- ksqlDB server `confluentinc/ksqldb-server:0.28.2`, pointed at `-Dksql.connect.url=…:8083`.

## 6. The data warehouse (destination)

- `internal/lmsdwh/` — the LMS analytics/reporting warehouse. **Star schema**: `dim_*` (dim_user,
  dim_student, dim_grade, dim_location, dim_course…) + `fact_*` (fact_student_enrollment,
  fact_assessment_result…). Entities `internal/lmsdwh/entities/*.go`; DDL e.g.
  `migrations/lmsdwh/1011_migrate.up.sql` (`dim_user` mirrors the ksqlDB join output). 77 migrations.
- **lmsdwh is itself a CDC source** (chained): `1001_migrate.up.sql:1-53` sets up
  `debezium_heartbeat` + `lmsdwh_dbz_signals` + `lmsdwh_publication`; a second source connector
  (`datawarehouses/lms/connectors/source/lmsdwh.json`) captures its dim/fact tables and re-emits them
  to per-partner DWHs (e.g. `lmsdwh_to_tokyo_jprep_dwh_dim_user.json`).
- **Accuracy verifiers** `internal/lmsdwh/internal/verify/` reconcile row counts; `hephaestus_dwh_accuracy`.

## 7. End-to-end trace — `bob.public.users` → `auth.users`

```
① row change in bob Postgres public.users
② WAL → slot local_manabie_bob, publication debezium_publication, plugin pgoutput
   source connector local_manabie_bob_source_connector (bob.json)
③ Debezium change event → topic local.manabie.bob.public.users  (Avro; route regex renames)
④ (warehouse path only: ksqlDB dim_user stream/join — plain sync skips this)
⑤ sink local_manabie_bob_to_auth_users_sink_connector (bob_to_auth_users.json):
   unwrap (extract new row) → route $5 → table "users" → JDBC upsert on pk user_id → auth_url
⑥ auth Postgres public.users
```

## Notable / bespoke (the "not vanilla" points)
- **Custom Java SMT** `CustomFilter.java` (repo-authored Transformation, passes row iff
  `filter.field == filter.value`).
- **Multi-tenant filtering at the sink:** partner DWH sinks use Confluent `Filter$Value` JSONPath on
  `resource_path` (`lmsdwh_to_tokyo_jprep_dwh_dim_user.json:12-15`).
- **Chained CDC** (lmsdwh is both sink and source — double Debezium hop).
- **Publications by SQL migration**, not Debezium (`autocreate.mode = disabled`).
- **Deletes intentionally dropped** (soft-delete via `deleted_at`).
- **Git-as-desired-state control plane** with a delete-guard (only JDBC sinks; never sources/ksqlDB).
- **Incremental-snapshot signal = direct SQL INSERT** into `dbz_signals` (NATS flow abandoned).
- **Semantics:** at-least-once + idempotent JDBC upsert on PK; no exactly-once/transactional config.
- **Auto-heal cron** every 10 min restarts FAILED/UNASSIGNED tasks (`cron_restart_failed_connector.yaml:45-55`).
- **Naming migration in flight:** literal `local_manabie_…` → Helm-templated; `PreWriteCheck` guards.