Lesson 7 · Kafka Connect & the sink · the repo signature

The connector-generation system

Nobody hand-writes 248 connectors. A few declarative YAML files fan out into hundreds — schema-aware, env-aware, org-aware.

Your win: explain how one small YAML pipeline definition generates hundreds of connector JSON files, and why the generator reads real database schemas to do it.

The problem: 248 connectors is a lot of JSON

You saw one sink config in Lesson 6. Now multiply: ~237 sink connectors, each a (source table → destination DB) pair, times every environment and organization. Hand-maintaining that is impossible — a column rename would mean editing dozens of files consistently. So this repo declares pipelines and generates the connectors.

The declarative definition

Each source database has one YAML file. The model is tiny — a source DB, the envs/orgs to generate for, and a list of pipelines (one source table → a set of destination databases):

internal/platform/datapipeline/datapipeline.go:12-30

type Def struct {           // one per source database
    SourceDatabase string
    Envs, Orgs     []string
    Pipelines      []*Pipeline
}
type Pipeline struct {      // one source table…
    SourceTable string
    Sinks       []SinkDecl    // …fanned out to these destination DBs
}

And here's a real one — bob's locations table needs to exist in twelve other services' databases:

deployments/helm/backend/hephaestus/generated_connectors/bob.yaml:1-19

source_database: bob
envs: [local]
orgs: [manabie, e2e]
pipelines:
  - source_table: locations
    sinks:
      - {database: calendar}   - {database: eureka}    - {database: fatima}
      - {database: auth}       - {database: lmsdwh}    - … (12 total)

The fan-out: env × org × table × sink

The generator (gen-data-pipeline) loads each def and multiplies it out — one connector JSON per (environment × organization × source table × sink):

datapipeline.go:106-128 (GenerateSinkFor, condensed)

for _, env := range d.Envs {
  for _, org := range d.Orgs {
    for _, p := range d.Pipelines {        // each source table…
      for _, sink := range p.Sinks {       // …to each destination DB
        SinkProperties{Env, Org, SourceTable, SinkDatabase}.Generate()   // → one connector JSON
}}}}

Do the multiplication That single locations pipeline — 12 sinks × 2 orgs × 1 env — is already 24 connector files. A source DB with dozens of tables, several orgs, and multiple envs reaches the hundreds fast. bob is the hub: 169 of the 237 sink connectors read bob's topics, because so many services need copies of bob's shared tables (users, locations, permissions). One bob.yaml, most of the pipeline.

Schema-aware: the generator reads real databases

This is the clever part. The generator doesn't just template strings — it loads the actual source and sink table schemas to compute two fields correctly:

Field	How it's derived	Why
`pk.fields`	the primary key, verified to exist on both sides	upsert needs a PK that both tables have
`fields.whitelist`	the intersection of source & sink columns	a column that exists in one but not the other would break the write

So if auth.users doesn't have a column that bob.users has, the generator simply leaves it out of that sink's whitelist. Two services can evolve their schemas independently and the generated connectors stay valid — the generation step absorbs the drift.¹

A guardrail during the migration There's a PreWriteCheck (sink.go:214-254) that, during the in-flight rename from literal names (local_manabie_…) to Helm-templated names, forbids changing a connector's name, topics, or field whitelist except in the exact way the migration allows. It's a generation-time safety net so a regeneration can't silently rewrite a live connector's identity or columns. Regenerate with gen-data-pipeline (or ./local/run.bash start rebuilds the dirs).

Configuration as data (and why generate it)

The generator is code; the connectors are data. Read the model and one generated file side by side — then the sink connector reference to see what each generated field means.

→ in-repo internal/platform/datapipeline/datapipeline.go + …/generated_connectors/bob.yaml
→ Confluent — JDBC sink config reference

Check yourself (from memory)

Q1. How many connectors does one source_table with 12 sinks, 2 orgs, 1 env produce?

1 env × 2 orgs × 1 table × 12 sinks = 24. The generator fans out every combination into its own JSON.

Q2. Why does the generator load real database schemas?

The whitelist is the intersection of source & sink columns, and the PK must exist on both — so independently-evolving schemas don't break the sink.

Q3. Where does the desired set of connectors actually live?

A few YAML defs → gen-data-pipeline → hundreds of JSON files in the repo. Those files are the desired state (Lesson 8).

Recall: the connector-generation system.

the model + fan-out + schema-aware, then reveal

Problem: ~237 sinks × envs × orgs — too many to hand-write. Declarative model (datapipeline.go): one Def per source DB = source_db + envs + orgs + pipelines: [{source_table, sinks:[{database}]}] (e.g. bob.yaml: locations → 12 destination DBs). Fan-out (gen-data-pipeline / GenerateSinks): env × org × table × sink → one connector JSON each (12 sinks × 2 orgs × 1 env = 24 files for one table). bob is the hub (169/237). Schema-aware: the generator loads real source & sink schemas → pk.fields (must exist on both) + fields.whitelist = intersection of columns → absorbs schema drift. PreWriteCheck guards the literal→Helm-template rename migration. Regenerate: gen-data-pipeline.

🎯 Interview one-liner "You have hundreds of connectors — how do you manage them?" → "We don't hand-write them. Each source database has a small YAML pipeline definition, and a generator fans it out across env × org × table × sink into hundreds of connector JSON files. The generator reads the real source and sink schemas to compute the primary key and the column whitelist as the intersection — so the two schemas can drift independently without breaking the sink."

Next: those generated files are just desired state. hephaestus is what makes the live cluster match them — the reconcile loop and the delete-guard that keeps a slot from ever being dropped. Ask me to work the fan-out math on a bigger def.

1. In-repo: internal/platform/datapipeline/datapipeline.go:12-136, sink.go:166-254, deployments/helm/backend/hephaestus/generated_connectors/bob.yaml.