60-Day Kafka 4 Learning Plan · Week 6 — Day 37 of 60

60-Day Kafka 4 Learning Plan · Week 6 — Kafka Connect & ksqlDB Sources: Kafka: The Definitive Guide Ch.9 · debezium.io/docs · docs.confluent.io/kafka-connectors/jdbc

Goal

Learn two ways to stream database changes into Kafka: the JDBC Source connector (SQL polling) and Debezium CDC (write-ahead log streaming). Understand when to use each, how to configure them, how the JDBC Sink connector writes topic data back to a database, and the operational gotchas specific to CDC — replication slot management, credential handling, and schema drift.

1. How JDBC Source connector works

The JDBC Source connector polls a database table at a configurable interval (poll.interval.ms). It detects new or changed rows using a tracking column strategy, then publishes each row as a JSON message to a Kafka topic.

PostgreSQL (orders table)
  → JDBC Source Connector (polls every 5 sec)
  → labs.db-orders (Kafka topic, JSON)

Offset tracking: stored in _connect-offsets topic
  - Last processed ID (incrementing mode)
  - Last processed timestamp (timestamp mode)

No DB schema changes needed — the connector queries existing tables via JDBC.

Required PostgreSQL table structure

CREATE TABLE orders (
    id          SERIAL PRIMARY KEY,        -- for incrementing mode
    order_code  VARCHAR(50) NOT NULL,
    amount      DECIMAL(10,2),
    status      VARCHAR(20),
    created_at  TIMESTAMPTZ DEFAULT NOW(),
    updated_at  TIMESTAMPTZ DEFAULT NOW()  -- for timestamp mode
);

-- Keep updated_at current on every UPDATE
CREATE OR REPLACE FUNCTION update_updated_at()
RETURNS TRIGGER AS $$
BEGIN NEW.updated_at = NOW(); RETURN NEW; END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER orders_updated_at
    BEFORE UPDATE ON orders
    FOR EACH ROW EXECUTE FUNCTION update_updated_at();

2. Polling modes — detect new rows

incrementing

Tracks a monotonically increasing ID column. Only detects INSERTs — UPDATE and DELETE are invisible.

"mode": "incrementing",
"incrementing.column.name": "id"

timestamp

Tracks an updated_at timestamp column. Detects INSERTs and UPDATEs — still no DELETE capture.

"mode": "timestamp",
"timestamp.column.name": "updated_at"

timestamp+incrementing ★ (recommended)

Combines both strategies. Detects INSERTs (via ID) and UPDATEs (via timestamp). Still no DELETE — use Debezium for that.

"mode": "timestamp+incrementing",
"timestamp.column.name": "updated_at",
"incrementing.column.name": "id"

Silent gap worth knowing: this mode relies entirely on the trigger in §1 correctly bumping updated_at on every write path. A bulk UPDATE run directly via a migration script, an ORM bulk-update that bypasses row-level triggers, or a manual psql fix that forgets to touch updated_at — all of these produce changes the connector will never see, with no error or warning. Polling-based CDC’s correctness is only as good as the application’s discipline around the tracking column, which is one of the reasons Debezium (§4) is generally preferred for anything where completeness matters.

3. JDBC Source connector config

# Deploy JDBC source connector
curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "labs-jdbc-source",
    "config": {
      "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
      "connection.url": "jdbc:postgresql://postgres:5432/labsdb",
      "connection.user": "labs",
      "connection.password": "secret",
      "mode": "timestamp+incrementing",
      "timestamp.column.name": "updated_at",
      "incrementing.column.name": "id",
      "table.whitelist": "orders",
      "topic.prefix": "labs.",
      "poll.interval.ms": "5000",
      "batch.max.rows": "100",
      "tasks.max": "1",
      "value.converter": "org.apache.kafka.connect.json.JsonConverter",
      "value.converter.schemas.enable": "true"
    }
  }'

This config has the exact schemas.enable=true trap flagged in Day 36 §7 — every row published by this connector carries a full embedded JSON schema alongside the actual data, which for a small table row (a handful of columns) can mean the schema envelope is larger than the row itself. For a table polled every 5 seconds at meaningful volume, that’s a real, avoidable bandwidth/storage cost. Set value.converter.schemas.enable: false for plain JSON, or better, switch to AvroConverter with Schema Registry — getting compatibility enforcement (Week 4) on the DB-derived schema for free.

Topic naming

topic.prefix + table name = topic name:

labs. + orders → labs.orders
labs. + users → labs.users

Install the plugin first

docker exec connect confluent-hub install confluentinc/kafka-connect-jdbc:10.7.4 --no-prompt
# Also install PostgreSQL JDBC driver
docker exec connect bash -c "cp /path/to/postgresql-42.7.3.jar /usr/share/java/kafka-connect-jdbc/"
docker-compose restart connect

4. Debezium CDC — capture ALL changes including DELETEs

Debezium reads directly from the PostgreSQL write-ahead log (WAL) via a replication slot. It captures INSERT, UPDATE, and DELETE events in real time — no polling, no schema changes, sub-second latency.

Enable WAL replication in PostgreSQL

-- postgresql.conf
wal_level = logical
max_replication_slots = 4
max_wal_senders = 4

-- Grant replication permission
ALTER USER labs REPLICATION;

Deploy Debezium PostgreSQL connector

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "labs-debezium-source",
    "config": {
      "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
      "database.hostname": "postgres",
      "database.port": "5432",
      "database.user": "labs",
      "database.password": "secret",
      "database.dbname": "labsdb",
      "database.server.name": "labs",
      "table.include.list": "public.orders",
      "plugin.name": "pgoutput",
      "slot.name": "labs_debezium_slot",
      "publication.name": "labs_publication",
      "transforms": "unwrap",
      "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
      "transforms.unwrap.drop.tombstones": "false"
    }
  }'

Debezium event envelope

{
  "before": { "id": 1, "status": "PENDING", "amount": 99.99 },
  "after":  { "id": 1, "status": "CONFIRMED", "amount": 99.99 },
  "op": "u",       // c=create, u=update, d=delete, r=read (snapshot)
  "ts_ms": 1721234567890,
  "source": { "db": "labsdb", "table": "orders", "lsn": 12345 }
}

Use the ExtractNewRecordState SMT (Single Message Transform) to unwrap the envelope and get just the after record for downstream consumers.

5. JDBC Sink connector — write topic back to DB

The JDBC Sink connector consumes from a Kafka topic and upserts rows into a target database table. It can auto-create and auto-evolve the target table based on the message schema.

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "labs-jdbc-sink",
    "config": {
      "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
      "connection.url": "jdbc:postgresql://analytics-db:5432/reports",
      "connection.user": "analytics",
      "connection.password": "secret",
      "topics": "labs.enriched-orders",
      "insert.mode": "upsert",
      "pk.mode": "record_key",
      "pk.fields": "order_id",
      "auto.create": "true",
      "auto.evolve": "true",
      "tasks.max": "2"
    }
  }'

Insert modes

6. JDBC Source vs Debezium CDC

7. Replication slot management — the WAL disk-bloat risk

Debezium’s “minimal DB load” advantage in §6 comes with a serious operational trade-off that’s easy to miss until it causes an incident: PostgreSQL cannot reclaim WAL segments that a replication slot hasn’t yet consumed. If the Debezium connector is paused, crashes, or is simply deleted without also dropping its replication slot, PostgreSQL keeps accumulating WAL on disk indefinitely for that slot — this is a well-known way to fill a production database’s disk and take it down entirely, unrelated to Kafka itself.

-- Check replication slot lag/retained WAL size — run periodically, not just when investigating
SELECT slot_name, active,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained_wal
FROM pg_replication_slots;

# If a connector is being permanently removed, its replication slot must be
# explicitly dropped — Kafka Connect does NOT do this automatically
curl -X DELETE http://localhost:8083/connectors/labs-debezium-source

-- Manual cleanup required after deleting the connector
SELECT pg_drop_replication_slot('labs_debezium_slot');

Why this deserves its own section, not just a bullet point: unlike most Kafka-side problems (which surface as consumer lag or a failed task, visible in the tools you already monitor), an orphaned replication slot is a PostgreSQL-side disk emergency that Kafka’s own monitoring won’t show you at all. Add pg_replication_slots retained-WAL monitoring to your DB alerting, not just your Kafka dashboards — and make “drop the replication slot” a mandatory step in any Debezium connector decommissioning runbook, not an afterthought.

8. Securing connection credentials

Every config example in this material (§3, §4, §5) includes "connection.password": "secret" in plaintext, submitted over the REST API and stored as-is in the _connect-configs topic and in Connect’s /connectors/{name}/config response — anyone who can read that topic or call that endpoint sees the real database password.

{
  "config": {
    "connection.password": "${file:/opt/kafka/secrets/db-creds.properties:db.password}"
  }
}

# Worker-level config — enable a ConfigProvider that resolves secrets at runtime
CONNECT_CONFIG_PROVIDERS: file
CONNECT_CONFIG_PROVIDERS_FILE_CLASS: org.apache.kafka.common.config.provider.FileConfigProvider

Combine with §11 from Day 36: Connect’s REST API needing authentication and connector configs needing externalized secrets are two halves of the same problem — even with REST auth in place, a plaintext password sitting in _connect-offsets/_connect-configs is readable by anyone with legitimate access to that internal topic (e.g. via kafka-console-consumer.sh), which is a wider blast radius than “who can call the REST API.” Use a ConfigProvider (file-based as shown, or a Vault/AWS Secrets Manager provider for production) so the actual credential never appears in the connector config JSON at all.

9. Monitoring CDC lag — different signals for polling vs streaming

“Lag” means something different for each connector type, and conflating them leads to the wrong diagnosis.

# Debezium exposes this directly — no need to compute it manually from WAL positions
# JMX: debezium.postgres:type=connector-metrics,context=streaming,server=labs
# Attribute: MilliSecondsBehindSource

Why the distinction matters operationally: a growing JDBC Source “lag” might just mean the poll interval is doing exactly what it’s configured to do (data genuinely isn’t visible until the next poll) — not necessarily a problem. A growing Debezium MilliSecondsBehindSource, on the other hand, means the connector is falling behind real database write volume, which combined with §7’s WAL retention risk is a much more urgent signal — a slow-consuming Debezium connector is simultaneously lagging and accumulating retained WAL on the source database.

10. Schema evolution when the source table changes

A column added to the orders table (say, discount_code) doesn’t require any Kafka Connect configuration change — the JDBC Source connector picks it up on its next poll, and Debezium picks it up from the next WAL event referencing it. But “the connector doesn’t error” isn’t the same as “downstream consumers handle it safely.”

JDBC Source with schemas.enable=true (or Avro/Protobuf): the schema embedded in each message changes the moment the DB schema changes — any downstream consumer expecting the old shape needs the same BACKWARD-compatibility discipline from Day 24, just triggered by a DB migration instead of an application code change.
Debezium: with Schema Registry integration, DB schema changes similarly produce a new registered schema version — subject to whatever compatibility mode (Day 24/26) is configured for that subject.
The real risk: a DB migration is usually reviewed by people thinking about the database impact (indexes, locking, migration runtime) — not necessarily by anyone thinking about the downstream Kafka consumer impact. A column rename or type change at the DB level is exactly the kind of “breaking schema change” Day 24 §3’s table warns about, but it can originate from a migration PR that never touches any .avsc/Kafka-related file at all.

Practical mitigation: if a CDC pipeline feeds consumers that matter, make schema-affecting DB migrations part of the same review process that schema changes get in Days 23–27 — a DB migration touching a CDC’d table should trigger the same compatibility-check thinking as a direct schema file change, even though nothing in the migration PR looks like a “Kafka change” on the surface.

11 Common pitfalls

Relying on a manual trigger for updated_at correctness — bulk updates or migration scripts that bypass the trigger create silent, undetectable gaps in JDBC Source polling coverage (§2)
Deploying with value.converter.schemas.enable: true for meaningful table volume — the same avoidable overhead flagged generally in Day 36 §7, concretely reproduced in the example config here (§3)
Deleting a Debezium connector without dropping its replication slot — the single most dangerous operational mistake in this material; causes unbounded WAL growth on the source database, unrelated to and invisible from Kafka-side monitoring (§7)
Leaving connection.password in plaintext in connector configs — readable by anyone with access to _connect-configs/_connect-offsets, a wider exposure than just REST API access (§8)
Treating a DB schema migration as unrelated to Kafka schema compatibility — a column rename/type change on a CDC’d table is a breaking schema change by any other name, just originating outside the usual schema-review process (§10)

Key Takeaways

JDBC Source polls tables — good for INSERT/UPDATE capture, no schema changes to DB, but its correctness depends entirely on the tracking column being reliably maintained
timestamp+incrementing mode ★ is the best JDBC polling strategy — catches both INSERTs and UPDATEs, but bulk operations bypassing the update trigger create silent gaps
Debezium reads the WAL — captures INSERT + UPDATE + DELETE with sub-second latency, at the cost of real replication-slot management responsibility
Deleting a Debezium connector requires manually dropping its PostgreSQL replication slot — otherwise WAL accumulates unboundedly and can fill the source database’s disk
Externalize connector credentials via a ConfigProvider — plaintext passwords in connector configs are readable via the internal _connect-configs topic, not just the REST API
“Lag” means different things for JDBC Source (poll interval) vs Debezium (WAL replication lag) — don’t conflate the two when diagnosing
DB schema migrations on CDC’d tables are Kafka schema changes in disguise — bring them into the same compatibility review discipline as Week 4’s direct schema changes
JDBC Sink upserts topic records into a target table — auto.create creates the table from schema
For production, prefer Debezium over JDBC polling — zero extra DB load, true CDC — but budget for replication slot operational ownership

Support me through GitHub Sponsors.

Thank you for Reading !! See you in the next post.

➡️ Day 38: HTTP Sink — POST Kafka events to REST endpoints

Resources

📘 Kafka: The Definitive Guide — Chapter 9 (Kafka Connect)
🌐 debezium.io/docs
🌐 docs.confluent.io/kafka-connectors/jdbc
🌐 PostgreSQL — Replication Slots

Link to Medium blog