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

hephaestus, the control plane

The generated JSON is just a wish. hephaestus is the job that makes the live cluster match it — and refuses to do the one destructive thing.

Your win: explain the git-as-desired-state reconcile loop, the create-or-update logic, and the delete-guard that makes running it safe — the design that turns 248 config files into a live pipeline.

hephaestus is a job, not a server

Despite being a "service," hephaestus doesn't serve requests — it only exposes a dummy health check. It's a job runner: reconcile jobs registered via bootstrap.RegisterJob, run as Helm hooks on deploy (Go-techstack course, Lesson 10). The core job is hephaestus_upsert_kafka_connect.

Git-as-desired-state

The reconcile: delete stale, then upsert everything

// ① DELETE: connectors live in the cluster but no longer on disk (live − onDisk)
deleteConnecter.DeleteConnectors(client, GenSinkConfigDir, customSinkDir, SourceConfigDir)   // + Slack alert

// ② UPSERT: every connector JSON in each config dir
upsert(client, SourceConfigDir)     // the Debezium sources
upsert(client, GenSinkConfigDir)    // the ~237 generated sinks
if DeployCustomSinkConnector { upsert(client, SinkConfigDir) }   // hand-written custom sinks

upsert(dir) reads every JSON file in the directory and applies it. Applying one connector is a GET-then-create-or-update:

old := GetConnector(name)              // does it exist? (404 → nil)
if old == nil { TryCreateConnector(c) } // absent → POST create
else          { TryUpdateConnector(old, c) }

// TryUpdateConnector: PUT only if the config actually changed…
if !anyChangeInConfig(old, c) { return nil }   // no diff → no-op (idempotent reconcile)

So re-running the job when nothing changed does nothing — it only touches connectors whose config genuinely differs. After a change it waits for IsConnectorReady (the connector and all its tasks RUNNING), and — for a source with a newly-added table — fires the incremental-snapshot signal from Lesson 3 (the --sendIncrementalSnapshot path). Reconcile, verify, back-fill: one loop.

The delete-guard — the safety that makes this OK

Deleting the wrong connector could be catastrophic (Lesson 2: dropping a source drops its slot). So the delete logic is deliberately narrow:

connectorType := old.Config["connector.class"]
if !found || connector == strings.ToUpper(connector) { return nil }   // ← skip ksqlDB (ALL-CAPS names)
if connectorType == "io.confluent.connect.jdbc.JdbcSinkConnector" {
    TryDeleteConnector(connector)     // ← ONLY delete JDBC sink connectors
}                                     //   (sources fall through → never deleted)

The deploy sequence & self-healing

On deploy, numbered Helm hooks run in order (deployments/helm/backend/hephaestus/templates/): 2 init ksqlDB migration → 3 upsert kafka connect (--deployCustomSinkConnector --sendIncrementalSnapshot) → 4 validate → 5 migrate ksqlDB → 6 sync data warehouse. And between deploys, an auto-heal cron runs every 10 minutes: it lists connector status, finds tasks in FAILED/UNASSIGNED, and restarts them — so a transient failure recovers without a human.

Check yourself (from memory)

1. In-repo: cmd/server/hephaestus/upsert_kafka_connect.go:43-157, internal/golibs/kafkaconnect/connector.go:84-352 (Upsert/Delete/TryUpdateConnector).