Lesson 6 · Kafka Connect & the sink

JDBC sink connectors

The other end of the pipe: read a change topic, flatten the envelope, and idempotently upsert the row into a destination database.

Your win: read a real sink connector config top to bottom — the upsert, the two SMTs, the dropped deletes — and explain what each line does.

One real sink, line by line

Here's the connector that carries bob's users into auth's database. It's short, and every line earns its place:

local/configs/services/kafka-connect-gen-sink-connector/bob_to_auth_users.json
connector.class = "io.confluent.connect.jdbc.JdbcSinkConnector"   // the sink type
topics          = "local.manabie.bob.public.users"              // the source's change topic
connection.url  = "${file:…:auth_url}"                          // → AUTH's Postgres (secret from file)
insert.mode     = "upsert"                                     // insert-or-update on PK conflict
table.name.format = "users"                                    // write into auth.users
pk.mode         = "record_value"   pk.fields = "user_id"       // the PK is a field in the value
auto.create     = "false"                                      // the table must already exist
fields.whitelist = "user_id,email,user_group,resource_path,deleted_at,…"  // only these columns
transforms      = "unwrap,route"                              // two SMTs, in order (below)
delete.enabled  = "false"                                     // deletes are NOT applied

The two SMTs: unwrap, then route

Remember the Debezium envelope (Lesson 4) has before/after/op? The sink wants a plain row, and the topic name is the five-part local.manabie.bob.public.users. Two Single Message Transforms fix both:

bob_to_auth_users.json:12-16 · sink.go:181-185
transforms.unwrap.type = "io.debezium.transforms.ExtractNewRecordState"   // envelope → just the "after" row
transforms.route.type  = "org.apache.kafka.connect.transforms.RegexRouter"
transforms.route.regex = "([^.]+).([^.]+).([^.]+).([^.]+).([^.]+)"
transforms.route.replacement = "$5"                              // local.manabie.bob.public.users → "users"

unwrap (Debezium's ExtractNewRecordState) flattens the envelope to the new row state — so the JDBC sink sees plain columns, not after.email. route strips the five-part topic down to $5 = the bare table name. Two small transforms turn "a change event on a namespaced topic" into "a row for table users."

Upsert: idempotent by design

Why insert.mode = upsert matters Kafka delivery is at-least-once — the same change event can be delivered more than once (a rebalance, a retry). A plain INSERT would then fail on a duplicate key or create a dup. Upsert (insert-or-update on the primary-key conflict, here pk.fields = user_id) makes re-applying a message a no-op — the row just ends up the same. That's what makes the whole pipeline safe without exactly-once machinery (Lesson 11).1

Deletes are dropped — on purpose

bob_to_auth_users.json:17,20-21
delete.enabled                         = "false"
transforms.unwrap.drop.tombstones      = "true"
transforms.unwrap.delete.handling.mode = "drop"

A hard delete in bob does not delete the row in auth — the delete event and its tombstone are discarded. Instead the platform soft-deletes: setting deleted_at is a normal column update, which flows through as an upsert like any other. This matches the codebase-wide never-hard-delete rule — and it means the sink never destroys data it can't get back.

Two more things worth knowing (1) The whitelist isn't hand-typed — fields.whitelist is the intersection of the source table's and the sink table's columns, computed by the generator from real DB schemas (Lesson 7). So two independently-evolving schemas can't break the sink. (2) No dead-letter queue — these generated sinks have no DLQ configured, so a poison message fails the task (which the auto-heal cron restarts, Lesson 8). A real gap worth naming in an interview.
Read this next

The JDBC Sink connector & primary keys

The sink's config reference — insert.mode, pk.mode/pk.fields, upsert semantics — and a deep-dive on getting primary keys right.

→ Confluent — JDBC Sink connector
→ Robin Moffatt — JDBC sink primary keys · in-repo internal/platform/datapipeline/sink.go:166-200

Check yourself (from memory)

Q1. What does the unwrap SMT do?

ExtractNewRecordState turns {before, after, op} into the plain after row. (Stripping the topic name is route's job.)

Q2. Why is the sink configured for upsert?

Upsert on the PK is idempotent — a duplicate delivery just rewrites the same row, so at-least-once is safe.

Q3. What happens to a hard delete in the source table?

delete.enabled = false + drop tombstones. Deletes are modeled as deleted_at updates, which upsert through normally.
Recall: the JDBC sink config, line by line.
type + upsert + 2 SMTs + deletes, then reveal
Sink = io.confluent.connect.jdbc.JdbcSinkConnector: reads a topic → writes a DB (connection.url, secret from file). Key config: insert.mode=upsert + pk.mode=record_value + pk.fields=user_id → idempotent (safe under at-least-once); table.name.format; auto.create=false; fields.whitelist = intersection of source & sink columns (generated). Two SMTs (in order): unwrap (ExtractNewRecordState — envelope → the after row) then route (RegexRouter, 5-part topic → $5 = bare table name). Deletes dropped: delete.enabled=false + drop.tombstones=true + delete.handling.mode=drop → soft-delete via deleted_at (a normal update). No DLQ (a gap — poison msg fails the task → auto-heal restart).
🎯 Interview one-liner "How does the change land in the destination?" → "A Confluent JDBC sink connector. Two SMTs first — unwrap flattens the Debezium envelope to the row, route derives the table name — then it upserts on the primary key. Upsert makes it idempotent under at-least-once delivery. Hard deletes are dropped; we soft-delete with a deleted_at column that flows through as a normal update."
Next: this connector wasn't hand-written — it was generated. How ~248 connectors come out of a few YAML files, schema-aware. Ask me about any config line above.

1. Confluent — JDBC Sink connector. In-repo: …/kafka-connect-gen-sink-connector/bob_to_auth_users.json, internal/platform/datapipeline/sink.go:166-200.