Kafka Connect and Data Integration

Data Engineering & Modern Data Platforms · Lesson 42 · Real-Time Data Engineering and Kafka

Lesson Overview

Purchase lines at TechLearningPro are insert-only. Data Quality Engineering made that a rule: a refund is a new line with line_type = 'refund', and the sale line stays. Finance wants those lines in the raw lake the same day they commit. A Python script used to do the extract. It stored a watermark in a file on the host that ran it. The host was replaced on a Thursday. The file went with it. The replacement script started from "the beginning of the table," duplicated a week of revenue into the landing prefix, and the morning check treated the duplicates as growth.

The log that should have held that watermark was already the platform's job. /learn/kafka is the primer for that log: a replicated, retained, ordered record per partition. This lesson does not re-teach brokers. It teaches the integration runtime that sits beside them. Apache Kafka Connect runs source connectors that pull from an external system and write topics, and sink connectors that read topics and write an external system. The framework, not the script's disk, stores how far a source has read. Workers can restart. A second worker can take a task. A record the sink cannot write can be sent to a dead-letter topic instead of vanishing in a log line nobody owns.

Connect is a good tool for the purchase extract because the table is append-only and the identity column increases. It is the wrong tool, in JDBC-poll form, for enrollment. Change Data Capture Fundamentals already showed why a query watermark misses hard deletes. That pipeline is lesson 46, on Debezium, which is itself a Connect source. The framework is the same. The capture mechanism is not.

learning.progress is already a topic. The application produces it. Structured Streaming with Spark reads it for the five-minute tile and keeps its cursor in a checkpoint, not in a consumer group. A Connect sink can land the same topic as objects for the morning restatement. Those are two readers. Their progress is not one number.

This lesson uses Apache Kafka's Connect framework (the 3.x worker model that 4.x kept: distributed mode, the REST API, single message transforms, and sink dead-letter queues from KIP-298). The JDBC source and the S3 sink used in the walkthrough are separate open-source connectors published outside the Apache Kafka tarball. Class names below are the ones those projects document. Confirm them against the connector build you deploy. Nothing here is a TechLearningPro-specific API.

Learning Objectives

By the end of this lesson you will be able to:

1. Place Connect in the path from an external system to a topic, and from a topic to a landing zone, without treating it as a stream processor.
2. Explain workers, connectors, tasks, converters, and single message transforms as separate layers.
3. Choose distributed mode for a pipeline whose offset must survive the death of one host.
4. Configure an insert-only PostgreSQL extract with the JDBC source's incrementing mode, and state which changes that mode cannot see.
5. Keep JDBC poll mode off enrollment, where deletes are part of the product.
6. Apply a mask and a lineage field with built-in transforms, and say what a transform is not allowed to become.
7. Route sink failures to a dead-letter topic with errors.tolerance and the dead-letter settings, and refuse that setting on the financial source.
8. Land a time-partitioned raw prefix that Module 5 can turn into a lakehouse table, and keep the topic as the replay source.
9. Read connector status from the Connect REST API and distinguish a failed task from consumer lag.
10. Hand off a contract: topic, key, offset store, dead-letter topic, raw prefix, and the statement that this extract is insert-only.

Prerequisites

• Kafka Delivery Semantics and Transactions and Kafka Consumer Groups and Offset Management: source offsets in Connect are not the same cursor as a consumer-group commit, and a sink is a consumer.
• Change Data Capture Fundamentals: query capture misses deletes. You need that failure mode in your head before you accept a JDBC source.
• Data Quality Engineering: purchase lines are insert-only, money is non-negative, payment tokens never land. Connect does not relax those rules.
• Structured Streaming with Spark: learning.progress already has a reader whose cursor is a checkpoint. A Connect sink beside it is a second reader.
• /learn/kafka for topics, keys, and retention. /learn/postgresql for the table the source reads. This lesson does not install a cluster for you.

Detailed Core Concepts

What the framework is for

A connector is a plugin that knows one external system. The framework knows Kafka. It starts tasks, passes configuration, calls the plugin, converts bytes, runs a short list of transforms, and records progress. It does not join streams, window events, or hold business state. That work is Spark, Flink, or Kafka Streams (lessons 32, 33, and 44). Teams that put a join inside a transform are hiding a processor where the next reader cannot see it.

Two directions exist.

A source connector polls or listens to an external system and gives the framework records. The framework converts them and produces them to Kafka. Progress is "how far I have read in the external system," stored by Connect, not by the external system.

A sink connector is given records the framework has consumed from Kafka. It writes them out. Progress is a Kafka consumer offset for that sink's group. If the write fails and you have asked for a dead-letter queue, the framework can produce the failed record to another topic and move the sink offset forward. The original topic still holds the bytes until retention deletes them.

Apache Kafka ships the framework, a file source and file sink that are demonstrations, and MirrorMaker 2, which is a Connect source for replicating topics between clusters. It does not ship a PostgreSQL connector or an object-store sink. Those are plugins you add on the worker's plugin.path. Using the file sink as the purchase landing would be a demo wearing a production name: it appends to one file on one worker's disk.

Workers, connectors, tasks

Standalone mode is one process. Offsets live in a local file (offset.storage.file.filename). That is the Thursday failure, with a logo. Use it to learn the REST-less config file on a laptop. Do not use it for purchase lines.

Distributed mode is one or more worker processes that form a group. Three internal topics, all compacted, are the cluster's memory:

They must be compacted. A delete-retention or a short cleanup policy on the offset topic is how a source "forgets" that it had reached line_seq 4_000_000 and reads the table again. Replication factor belongs at the same bar as the rest of the cluster, which lesson 39 set. A replication factor of 1 on the offset topic makes the integration cluster weaker than the data topics.

A connector is the named configuration: class, tables, topic, transforms, error policy. A task is the thread of work. tasks.max is a ceiling. The connector decides how many tasks it can use. A JDBC source that reads one table in incrementing mode often runs one task, because the split is a single increasing key. Setting tasks.max to 12 does not create twelve independent snapshots of the same key, and it does not make the database twelve times happier. A sink's tasks are assigned topic partitions. Extra tasks beyond the partition count sit idle. The ceiling that matters for the S3 sink is the partition count of the topic you are landing, which lesson 39 sized for the producers, not for the sink.

Converters sit on the Kafka boundary

Inside the framework, a record has a schema and a value (Connect's data model). On the topic, a record is bytes. A converter is the only component that crosses that boundary.

org.apache.kafka.connect.json.JsonConverter with schemas.enable=true writes and reads a JSON envelope that includes a schema and a payload. It is not "a JSON object of the row." schemas.enable=false writes the payload as ordinary JSON. Spark's from_json, and any consumer that expects a bare object, needs the false setting if the topic is supposed to be plain JSON. Mixing them is a classic outage: the sink expects an envelope, the producer wrote a bare object, and every record fails conversion.

org.apache.kafka.connect.storage.StringConverter treats the whole value as a string. org.apache.kafka.connect.converters.ByteArrayConverter passes bytes through and is what you want when the payload is already Avro or Protobuf and Connect must not reinterpret it. Avro that uses a schema id from a registry is the next lesson's converter (io.confluent.connect.avro.AvroConverter and the registry URL). Do not point the purchase topic at that converter until lesson 43 has a subject and a compatibility level.

Worker-level key.converter and value.converter are defaults. Connector-level settings override them. The purchase source and the progress sink may legitimately differ. Set the converter on the connector so a worker default change does not rewrite one pipeline by accident.

Transforms are single-message and optional

A single message transform (SMT) changes one Connect record at a time, in the worker, after a source connector returns it or after a sink converter has parsed it. The built-in types live in org.apache.kafka.connect.transforms and are part of Apache Kafka. The ones this lesson uses:

• MaskField$Value replaces named fields with the empty value of their type (empty string, zero, false). It is a coarse mask, not a hash and not a redaction policy. Payment tokens should not be selected by the source at all. Mask is the backstop when a column arrives anyway.
• InsertField$Value can copy the topic name, partition, and offset onto the value. Those fields are how a raw file points back at the record that produced it.
• RegexRouter rewrites the topic name. It belongs on a source whose default topic naming is noisy, and it is easy to use as a silent fork. Review it.
• ValueToKey sets the Kafka key from value fields. The purchase topic's key should be the line identity, not a null key, so log compaction and downstream deduplication have something to hold.

Transforms run in the order listed. They see one record. They do not see the previous record, the table, or the clock beyond what InsertField can stamp. A "transform" that calls a warehouse, waits on a dimension lookup, or drops records based on a remote allow-list will stall the task and hide a dependency. That logic belongs in a processor or in the curated job, after raw has landed.

Filter exists and can drop records. Dropping a purchase line in the source task is a silent gap in revenue. Do not filter business rows in the extract. Land them. Let the quality gate in lesson 17 quarantine.

Dead-letter queues are a sink feature

KIP-298 added error handling that the sink path can use. The settings are framework configuration, not a custom plugin:

• errors.tolerance = none or all. none fails the task. all skips the record after the configured reaction.
• errors.deadletterqueue.topic.name is the topic that receives the failed record.
• errors.deadletterqueue.context.headers.enable = true adds headers for the error, the original topic, and the connector. Turn it on. A dead-letter topic without the reason is a second junk drawer.
• errors.deadletterqueue.topic.replication.factor must be set if the framework is allowed to create the topic.
• errors.log.enable and errors.log.include.messages log the failure. Logging is not a queue. Disks rotate.

The dead-letter topic receives records the sink could not convert, transform, or deliver, depending on where the exception was thrown. It does not receive a row the JDBC source never managed to read. Source-task failures still fail the task when you leave tolerance at none. That is what you want for purchase lines: a poison row stops the extract, the offset stays put, and someone looks at the row. Tolerance all on a source can move the stored position past a row that never reached Kafka, and the framework will not have a sink-style dead letter for that gap.

When a sink dead-letters a record, the sink offset moves past it. The original bytes remain on the source topic until retention. Replay is: fix the reader, then reconsume from the original offset or reprocess the dead-letter topic. Deleting the dead-letter topic without a count is how you lose the only list of what you skipped.

JDBC incrementing mode is a poll

The Confluent JDBC source connector (io.confluent.connect.jdbc.JdbcSourceConnector, distributed separately from Apache Kafka) reads tables with SQL. mode=incrementing records the greatest value it has seen of incrementing.column.name and next time asks for values strictly greater than that. mode=timestamp uses a timestamp column and has the tie and clock problems lesson 16 spelled out for updated_at. mode=timestamp+incrementing uses both, for tables that update. mode=bulk rereads the whole table on each poll.

Purchase lines are inserts. Give them a bigint identity column that the application never updates, and use incrementing mode. The connector will not see an UPDATE or a DELETE. That is acceptable only because the product forbids both. Revoke UPDATE and DELETE from the application role so the technical limitation and the business rule are the same rule.

enrollment is updated and deleted. Incrementing mode would keep unenrolled learners in the lake forever. Timestamp mode would miss a delete and any update that does not move the timestamp. The extract for that table is the log reader in lesson 46.

What lands is a raw prefix, not a lakehouse table

The S3 sink writes objects. A time-based partitioner can put them under YYYY/MM/dd/HH in UTC. Those objects are a landing zone: immutable enough to replay, small enough to be annoying, and not a table with a snapshot. Module 5, starting at Data Lake Architecture and Data Lakehouse Architecture, is where a commit, a schema, and a reader contract turn a prefix into something analytics can query. This lesson's success criterion is: every committed purchase line reaches the topic, and the sink either writes it into the prefix or into the dead-letter topic, with the Kafka offset recoverable.

Flush and rotation decide file sizes. flush.size is a record count. rotate.interval.ms closes a file on a time boundary derived from the record timestamp you configured. A huge flush and a long rotate makes recovery coarse. A flush of one record makes a small-file problem the morning job will pay for. Pick a size the batch readers in lesson 31 can open without spending the slot on file listings, and write the number in the connector config review.

Architecture and Design Diagrams

mermaidCopy
flowchart LR
    subgraph pg [PostgreSQL]
        lines[purchase_line insert-only]
    end

    subgraph connect [Connect distributed worker]
        srcTask[JDBC source task]
        srcConv[Value converter]
        srcSmt[MaskField and InsertField]
        sinkTask[S3 sink task]
        dlqProd[Dead-letter producer]
    end

    subgraph kafka [Kafka]
        topic[commerce.purchase_line]
        offsets[(offset.storage.topic)]
        dlq[commerce.purchase_line.dlq]
    end

    subgraph landing [Raw landing]
        prefix[UTC hourly prefix]
    end

    lines --> srcTask
    srcTask --> offsets
    srcTask --> srcConv --> srcSmt --> topic
    topic --> sinkTask
    sinkTask --> prefix
    sinkTask -->|conversion or delivery failure| dlqProd --> dlq

The offset topic holds the greatest line_seq the source has produced. The S3 sink's progress is a consumer-group offset on commerce.purchase_line. Those cursors move independently. A sink that is an hour behind does not mean the source has stopped reading PostgreSQL.

mermaidCopy
sequenceDiagram
    participant DB as purchase_line
    participant Src as JDBC source task
    participant Off as Offset topic
    participant Topic as commerce.purchase_line
    participant Sink as S3 sink task
    participant S3 as Raw prefix
    participant DLQ as Dead-letter topic

    Src->>DB: SELECT rows with line_seq greater than stored position
    DB-->>Src: Next batch of inserts
    Src->>Topic: Produce with key order_id and line_id
    Src->>Off: Commit new line_seq after produce
    Sink->>Topic: Consume next offsets
    alt Record converts and the object write succeeds
        Sink->>S3: Append to the UTC hour prefix
        Sink->>Topic: Commit sink offset
    else Record fails conversion or the object write
        Sink->>DLQ: Produce original bytes and error headers
        Sink->>Topic: Commit sink offset past the bad record
    end

A source failure before produce does not take the dead-letter branch. The task stops. The stored line_seq stays at the last successful commit. The next start reads from there.

Technical Explanation

Distributed worker settings that are the cluster

The worker configuration is the process, not the connector. The properties that define the cluster:

propertiesCopy
bootstrap.servers=kafka-1:9092,kafka-2:9092,kafka-3:9092
group.id=connect-tlp
config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-status
config.storage.replication.factor=3
offset.storage.replication.factor=3
status.storage.replication.factor=3
plugin.path=/opt/connectors
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false

group.id is the Connect cluster's coordination group. It is not the sink's consumption group. Create the three topics compacted, or let the worker create them with the replication factors above on a cluster that allows it. plugin.path must contain the JDBC connector and the S3 connector as installed by their own distributions. Putting jars in a random directory and restarting one worker is how tasks fail with ClassNotFoundException on the worker that did not get the file.

Credentials use a config provider so the password is not stored in connect-configs as plain text. Apache Kafka's FileConfigProvider is the long-standing mechanism:

propertiesCopy
config.providers=file
config.providers.file.class=org.apache.kafka.common.config.provider.FileConfigProvider

Connector values then look like ${file:/var/run/secrets/jdbc.properties:password}. The file is on the worker, with permissions limited to the Connect user. Rotating the database password means updating the file and restarting the task, which is an operational step you write down. It is still better than a password in a compacted topic that replicas will hold until you compact it out.

Source offsets versus sink offsets

For the JDBC source, the offset record is the connector's memory of line_seq. You can inspect it through the Connect REST API added for offset management (Kafka 3.5 and later): GET /connectors/{name}/offsets while you understand the shape, and DELETE /connectors/{name}/offsets only when the connector is stopped and you intend a reread. On an older worker, the same data lives in connect-offsets and you do not edit that topic by hand.

For the S3 sink, progress is a consumer group. The group id is derived from the connector name (commonly connect-<name>). Confirm the actual id with kafka-consumer-groups.sh --list on your build rather than assuming a prefix from a blog. Lag on that group means the prefix is behind the topic. Lag of zero on a group that does not exist means you looked up the wrong id. Spark's tile, reading the same progress topic in another pipeline, will not appear in this list as a Connect group. Its cursor is the checkpoint from lesson 32.

Header lineage

InsertField can attach the Kafka coordinates to the value that lands in the object. That makes a support question answerable: which offset produced this Parquet row. It does not replace the topic. If the object store loses a file, you replay the topic. If the topic has aged out, the object is the surviving copy and you cannot reconstruct the hour from Kafka. Retention on commerce.purchase_line has to cover the time you are willing to reprocess, including a sink outage. Lesson 39's retention discussion is the control. This lesson's requirement is: retention longer than the worst sink outage you will recover by replay, plus a margin.

Idempotency of the landing

The source produces at least once if a produce succeeded and the offset commit did not. A restart resends from the stored line_seq. Consumers of the topic, including the S3 sink, may see a duplicate line_seq. The table's unique (order_id, line_id) and the identity column are the keys a curated job uses to collapse duplicates. The raw prefix is allowed to contain a duplicate row for that reason. A curated publish that sums raw files without deduplicating on (order_id, line_id) will invent revenue. That dedupe is the batch job's responsibility, and it is the same uniqueness check lesson 17 already required. Connect did not add a new exactly-once promise across PostgreSQL and S3.

Sink delivery to object storage is not a Kafka transaction. A retried upload can leave two objects that share rows. Design the curated read to be correct in the presence of overlapping raw files: dedupe on the business key, and treat the raw prefix as append-only evidence.

Practical Implementation or Design Walkthrough

Problem

Land every insert into purchase_line on commerce.purchase_line, then into s3://tlp-raw/commerce/purchase_line/. Keep payment fields out. Keep a dead-letter topic for the sink. Do not capture enrollment with this connector.

Table

sqlCopy
CREATE TABLE purchase_line (
    line_seq bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    order_id text NOT NULL,
    line_id text NOT NULL,
    course_id text NOT NULL,
    line_type text NOT NULL CHECK (line_type IN ('sale', 'refund')),
    unit_price_cents bigint NOT NULL CHECK (unit_price_cents >= 0),
    quantity integer NOT NULL CHECK (quantity > 0),
    event_time timestamptz NOT NULL,
    UNIQUE (order_id, line_id)
);

REVOKE UPDATE, DELETE ON purchase_line FROM app_commerce;

line_seq is the poll cursor. (order_id, line_id) is the business key. event_time is the business timestamp in UTC, used later for the hourly prefix. There is no email column and no payment token column. Do not add them "for the sink to mask."

Source connector

Submit this to a distributed worker. The URL is the Connect REST API (POST /connectors), which Apache Kafka documents.

jsonCopy
{
  "name": "purchase-line-jdbc",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "connection.url": "jdbc:postgresql://pg-commerce:5432/commerce",
    "connection.user": "connect_reader",
    "connection.password": "${file:/var/run/secrets/jdbc.properties:password}",
    "table.include.list": "public.purchase_line",
    "mode": "incrementing",
    "incrementing.column.name": "line_seq",
    "topic.prefix": "commerce",
    "poll.interval.ms": "60000",
    "tasks.max": "1",
    "numeric.mapping": "best_fit",
    "db.timezone": "UTC",
    "transforms": "pickKey,lineage,maskUnexpected",
    "transforms.pickKey.type": "org.apache.kafka.connect.transforms.ValueToKey",
    "transforms.pickKey.fields": "order_id,line_id",
    "transforms.lineage.type": "org.apache.kafka.connect.transforms.InsertField$Value",
    "transforms.lineage.topic.field": "kafka_topic",
    "transforms.maskUnexpected.type": "org.apache.kafka.connect.transforms.MaskField$Value",
    "transforms.maskUnexpected.fields": "payment_token,email",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "errors.tolerance": "none"
  }
}

topic.prefix=commerce plus the table name yields a topic derived from the table. Confirm the actual topic in GET /connectors/purchase-line-jdbc/topics or by listing topics after the first poll, and align the sink to that name. Older JDBC builds used table.whitelist for the same purpose as table.include.list. Use the property your connector version documents.

MaskField on payment_token and email is a backstop. Those columns are not in the table. If someone adds them, the field is emptied in the topic payload. The task still loads the column from PostgreSQL, so the backstop is not a substitute for keeping the column off the table. errors.tolerance stays none.

poll.interval.ms of 60 seconds is a freshness choice. Purchase lines do not need sub-second visibility. A one-second poll on a table this size mostly wakes the database. The cohort product's fifteen-minute bound is a different pipeline.

The reader role has SELECT on purchase_line only. It does not have REPLICATION. That privilege appears in lesson 46 and would be a surprise on a poller.

Sink connector

jsonCopy
{
  "name": "purchase-line-s3",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "topics": "commerce.purchase_line",
    "s3.bucket.name": "tlp-raw",
    "s3.region": "us-east-1",
    "topics.dir": "commerce/purchase_line",
    "storage.class": "io.confluent.connect.s3.storage.S3Storage",
    "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "path.format": "'year'=YYYY/'month'=MM/'day'=dd/'hour'=HH",
    "partition.duration.ms": "3600000",
    "locale": "en-US",
    "timezone": "UTC",
    "timestamp.extractor": "RecordField",
    "timestamp.field": "event_time",
    "flush.size": "10000",
    "rotate.interval.ms": "600000",
    "tasks.max": "3",
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "commerce.purchase_line.dlq",
    "errors.deadletterqueue.context.headers.enable": "true",
    "errors.deadletterqueue.topic.replication.factor": "3",
    "errors.log.enable": "true",
    "errors.log.include.messages": "false"
  }
}

Class names are the Confluent S3 sink's. errors.log.include.messages=false avoids writing purchase payloads into worker logs. The dead-letter topic holds the payload for people who are allowed to see it. flush.size and rotate.interval.ms are starting points for a low-volume financial table: close files often enough that an hour is not one giant part, and rarely enough that you are not creating an object per poll. Measure object counts after a day and adjust. tasks.max should not exceed the topic's partition count.

timestamp.extractor=RecordField means the prefix follows event_time, not the worker's wall clock. A line that commits late still lands in the hour it happened. The morning job that reads "yesterday" must use event_time, not object creation time, or a late file in an old hour will be missed. That is the same event-time discipline as the tile.

Validation

1. Insert two lines. Within two poll intervals, both keys are on the topic. Offsets for purchase-line-jdbc show the new line_seq.
2. Restart the worker. Insert another line. The connector does not resend the first two unless the produce/commit boundary duplicated them. A duplicate, if you force one by failing a commit, collapses in a SELECT DISTINCT on (order_id, line_id) check you run on the raw data.
3. Delete or update a row as a superuser in a lab. The topic does not change. That is the documented gap. The revoke is what stops it happening from the application.
4. Point the sink at a canary topic with one malformed record and errors.tolerance=all. The dead-letter topic receives it with error headers. The good records are in the prefix. Task status stays RUNNING.
5. Set the source to a bad password. Task status goes FAILED. line_seq does not jump. Fix the secret and POST /connectors/purchase-line-jdbc/restart?includeTasks=true.

Status comes from GET /connectors/purchase-line-jdbc/status and the same path for the sink. The field tasks[].state is the one on-call reads. FAILED with a trace is a stopped extract. A RUNNING sink with growing group lag is a slow landing, which is a different page.

What you do not deploy in this lesson

No connector on enrollment. No transform that writes to a warehouse. No claim that the S3 objects are an Iceberg or Delta table. The prefix is the handoff. Module 5 registers a table on top of a reviewed layout. Lesson 47 puts lag and dead-letter rate on a dashboard. Lesson 43 replaces bare JSON with a registry-backed contract if this topic needs to evolve in place.

Enterprise Real-World Scenarios

Purchase lines into the raw lake

Commerce commits a few tens of thousands of lines on a peak day, and a spike when a promotion runs. Freshness for finance is "yesterday's lines are queryable in the morning," which the hourly prefix satisfies with margin. The stricter need is completeness: a missing hour is a revenue hole, a duplicated hour is a revenue fiction. Incrementing mode plus a revoked UPDATE/DELETE, a dead-letter topic on the sink, and a curated dedupe on (order_id, line_id) meet that need. Cost is one small connector task, a poll every minute against an indexed line_seq, and object-store writes bounded by flush.size. Security is a SELECT-only role, a password file on the worker, and no payment columns in the SELECT list. The business decision this supports is the morning revenue number. It does not support a live authorization check. That stays on the operational database.

The same progress topic, two readers

learning.progress already feeds the Spark tile. The batch completion job wants files, not a broker connection, because the restatement window in lesson 32 rereads landed data. A second Connect sink, progress-s3, consumes learning.progress in its own group and writes s3://tlp-raw/learning/progress/. The Spark checkpoint does not move when this sink lags, and this sink's lag does not mean the tile is down. A team once added errors.tolerance=all on that sink and left the dead-letter name blank. Malformed events were logged and skipped. The tile, which fails the micro-batch on bad input unless you coded otherwise, and the batch job, which never saw the skipped offsets, diverged by the dead-letter count nobody stored. The fix was the dead-letter topic plus a page on its produce rate. The tile's definition did not change. Two readers, two cursors, one topic, and a place for the rows one reader refused.

A third request often shows up in the same design review: "also filter progress down to one course for a partner extract inside the connector." That is a new consumer with its own group, or a curated table with a predicate. It is not a Filter transform on the raw sink. A filter on the raw sink changes the landing everyone else will eventually trust.

Best Practices

• Run integration pipelines in distributed mode. Treat standalone as a local exercise.
• Compact the three internal topics and replicate them like the data topics. Alert if their cleanup policy changes.
• One connector, one direction, one system. A JDBC source that also "cleans" data is two jobs sharing a process.
• Set converters on the connector. Plain JSON for this lesson's purchase topic means JsonConverter with schemas.enable=false.
• Keep source errors.tolerance at none when a skipped row is a financial or enrollment hole. Use all with a dead-letter topic on sinks, and turn context headers on.
• Put the business key on the Kafka record with ValueToKey when the source does not set a key.
• Stamp topic, partition, and offset onto the landed record if the sink format can carry them, so an object can be traced to a log position.
• Store secrets through a ConfigProvider. Restrict the worker's file permissions. Do not commit connector JSON with a live password into git.
• Size flush.size and rotation for the readers you have. Count objects after the first real day.
• Write the insert-only assumption next to the JDBC mode. Revoke the statements the mode cannot see.
• Keep raw prefixes and dead-letter topics in the ownership of the ingest squad, with retention and a replay note. Curated dedupe stays downstream.
• Do not use Connect transforms as the quality gate from lesson 17. The gate runs before curated publish, where severities exist.

Common Mistakes and Anti-Patterns

Standalone offsets for a shared pipeline. The offset file dies with the host. The next start rereads or skips, and both look like a green task. Use distributed mode and the offset topic.

JDBC timestamp mode on a table that deletes. The poll never returns the missing key, so the lake keeps it. This is the enrollment bug with a connector class name. Use a log reader, or restrict the table to inserts and use an incrementing key.

errors.tolerance=all without a dead-letter topic. The sink logs and moves on. Retention eats the original context. You are left with a rotated log file. Name the dead-letter topic and alert on it.

schemas.enable left at the worker default. One connector writes envelopes and another writes bare JSON. Downstream from_json fails closed or, worse, parses the envelope as if the schema document were the event. Set the flag on each connector and test one record.

A transform that filters or joins. Business drops in the source task never reach the raw prefix, so the curated gate cannot quarantine them. It can only notice that the count is short, if anyone reconciles. Land the row. Join in the processor or the warehouse.

tasks.max treated as throughput. Twelve JDBC tasks on one incrementing key means one task works. Twelve sink tasks on a three-partition topic means three tasks work. Raise partitions with a reason from lesson 39, then raise tasks to match.

Editing the offset topic to "fix" a rewind. Compacted offset records are the cursor. A hand-produced offset is an untested rewrite of history. Stop the connector and use the offsets API on a worker that has it, with a written plan for duplicates.

Calling the S3 prefix a table. Analysts point a query engine at the prefix, the hour is incomplete because the sink has not rotated, and a number escapes. The prefix is raw. The table is a Module 5 commit with an owner.

Masking as permission to select secrets. MaskField empties a field after the worker has read it. The database, the worker heap, and an error log with include.messages=true still saw it. Do not select the column.

Troubleshooting and Design Review

Symptom: The task is RUNNING and the topic stopped growing. PostgreSQL has new line_seq values.

Likely cause: the stored incrementing offset is ahead of the rows you think are new, or the poll is failing in a way the status does not surface yet, or you are looking at the wrong topic name after topic.prefix.

Investigation: GET /connectors/purchase-line-jdbc/status and GET /connectors/purchase-line-jdbc/offsets. Compare the stored value with SELECT max(line_seq) FROM purchase_line. Confirm the topic the connector actually writes.

Corrective action: if the offset is ahead because of a bad manual edit, stop and reset with a plan. If the offset is behind and the task is running, read the worker log for query errors and connection-pool exhaustion. Do not insert a dummy row to "kick" it until you know which of those it is.

Symptom: After a worker replacement, the first poll reloads millions of lines.

Likely cause: a new group.id, a new offset.storage.topic, or an offset topic that was deleted and recreated empty. The connector correctly believes it has never read the table.

Investigation: the worker config on the new host, and whether connect-offsets has records for this connector name.

Corrective action: point the worker at the existing internal topics. If the offset is truly gone, choose a starting line_seq on purpose (the offsets API or a new connector name plus a documented overlap) and dedupe downstream. Do not let a full reread land into a curated table that sums without a key.

Symptom: The dead-letter topic is growing and the prefix looks complete in the dashboard.

Likely cause: tolerance all is doing what you asked. The dashboard counts objects, not records versus line_seq.

Investigation: dead-letter produce metrics and a count of dead-letter records against the source-topic rate. Read one header before you read a hundred payloads.

Corrective action: fix the converter or the record. Replay the dead-letter topic through a corrected sink, or reconsume the original offsets. Add the dead-letter count to the landing check so "prefix moved" is not the success test.

Symptom: Hour directories in S3 are shifted by five hours.

Likely cause: timezone is the worker default or a local zone, or timestamp.extractor is Wallclock and the workers are not on UTC.

Investigation: one object's path against the event_time inside it.

Corrective action: set timezone=UTC and timestamp.extractor=RecordField. Rewrite or repartition the shifted hours once, as a batch, and keep the topic so you can do that from evidence.

Symptom: Spark's progress tile is healthy and the progress landing prefix is hours behind, or the reverse.

Likely cause: you compared two different cursors and paged the wrong team.

Investigation: the Connect consumer group lag for the sink, and the Spark query's last committed offset in its checkpoint. They are allowed to differ.

Corrective action: page the owner of the cursor that is late. Do not restart the healthy one.

A design review asks: Is this table insert-only, and where is that enforced? Where is the source offset stored, and who can delete it? What is the sink group id? What happens to a record the sink cannot write? Which prefix is raw, and which job is allowed to publish a number? Does any transform drop rows? Is enrollment on this connector? If the answer to the last question is yes, the review stops and the design moves to lesson 46.

Interview Questions and Answers

Q1. What does Kafka Connect do that a Python poller with a local watermark file does not?

Connect runs a connector plugin inside a worker that stores source progress in a replicated, compacted Kafka topic, and sink progress as a consumer offset. A second worker can pick up the task without copying a file off a dead host. Converters and single message transforms are configured at the edge of the topic, and a sink can send failures to a dead-letter topic. The poller can be written correctly, with the watermark in a durable store and with idempotent loads. Most of the ones that fail were durable only on the host that ran them. I use Connect when the extract is a standard source or sink and the offset has to be operated like the rest of the log. I do not use it to hide a join.

Q2. How do a connector, a task, a converter, and a transform differ?

The connector is the plugin and its configuration for one external system. Tasks are the units of work it splits into, up to tasks.max. The converter turns bytes on the topic into Connect data and back. The transform changes one Connect record in the worker and does not talk to Kafka by itself. I can restart tasks without changing the converter, and I can change a mask without rewriting the JDBC poll. Mixing those layers is how a password ends up in a transform jar that also contains business logic nobody can replay.

Q3. Why does JsonConverter with schemas.enable=true break a consumer that expects a plain object?

That setting writes a JSON document with a schema and a payload, which is Connect's envelope, not the row. A consumer or Spark from_json schema written for order_id and line_id at the top level will not find them. schemas.enable=false writes the payload alone. I set the flag on the connector, produce one record, and read it back with the same decoder the downstream job uses. I do not inherit the worker default and assume.

Q4. When is a dead-letter topic actually used?

On a sink, when errors.tolerance=all and errors.deadletterqueue.topic.name is set, records that fail in the sink path are produced there and the sink offset moves on. Context headers should be enabled so the error is stored with the bytes. I do not get that behavior from a JDBC source task that cannot read the database. For purchase lines I leave the source on errors.tolerance=none so a bad row stops the task before the offset moves past it. Tolerance without a dead-letter name is a skip. I treat that as a defect in the config review.

Q5. Where does the JDBC source remember its position, and what is that position for incrementing mode?

In distributed mode the position is a record in offset.storage.topic, keyed by the connector and the source partition it defines. For incrementing mode the value that matters is the greatest line_seq successfully handed to the framework. It is not a Kafka consumer offset, and it is not event_time. Losing that topic, or starting the worker with a different one, makes the connector believe it has never read the table. I inspect offsets through the Connect REST API on Kafka 3.5 and later, with the connector stopped before any reset, and I plan the duplicate load that a rewind creates.

Q6. Why is incrementing JDBC mode acceptable for purchase lines and unacceptable for enrollment?

Purchase lines are insert-only, with a new identity value on every line, and refunds are new rows. The mode reads values greater than the stored identity. It will not see updates or deletes, and the application role is revoked from issuing them. Enrollment rows change and disappear. A poll cannot report a key that is gone, which is the failure lesson 16 walked through. Enrollment uses a logical-decoding connector in lesson 46. I do not "start with JDBC and upgrade later" on a table whose deletes are the product.

Q7. How do you decide tasks.max?

I read it as a ceiling, then I ask what the connector can split. One incrementing table is one task. A sink can use one task per topic partition, and extra tasks idle. Raising tasks.max above the partition count does not increase sink parallelism. If the sink is slow because three partitions are not enough, that is a topic-design change with a key and an ordering consequence, not a task setting.

Q8. What exactly-once statement can you make about this database-to-object path?

I can say the source will retry from the last stored line_seq, so a crash around the commit can duplicate a row on the topic, and the sink can upload overlapping objects if a request is retried. I cannot say each business line appears once in the prefix. The curated job dedupes on (order_id, line_id) the way lesson 17 already required for uniqueness. Kafka transactions, from lesson 41, do not wrap the PostgreSQL commit and the S3 PUT in one transaction. I do not tell finance the raw prefix is a ledger.

Q9. How should secrets appear in connector configuration?

Through a ConfigProvider, so the compacted config topic stores a reference such as a file placeholder rather than the password. The file is readable by the Connect user and not by the repository. I still assume the worker can see the secret at runtime, which means task logs must not print records or connection URLs at debug in production. Masking a column is a separate problem from authenticating to the database.

Q10. What do you hand Module 5, and what do you refuse to claim?

I hand the topic name, the key, the JSON shape, the raw bucket prefix, the partition path in UTC, event_time as the path's timestamp, the dead-letter topic, retention, and the insert-only rule. I refuse to claim the prefix is a lakehouse table, that object creation time equals event time, or that a query engine pointed at the prefix is a supported interface. The table, the snapshot, and the grants are Module 5. The ingest squad owns replay from the topic while retention holds.

Q11. A partner asks for a filtered copy of the raw purchase sink. Where does that filter go?

On a consumer of the topic or on a curated table, with its own offset and its own owner. It does not go in a Filter transform on purchase-line-s3. That sink is the raw landing. A filter there changes the evidence the morning job is supposed to trust, and the dropped lines never reach a quarantine the quality gate can count.

Hands-On Exercises

Exercise: An insert-only extract with a sink dead letter

Problem statement. Stand up the purchase-line path against a lab PostgreSQL and a lab Kafka. Prove that a worker restart does not reread from line_seq 1, that an update is invisible, and that a sink failure is a dead-letter record rather than a lost offset.

Tasks.

1. Create purchase_line as specified, with UPDATE and DELETE revoked from the reader and the application role.
2. Run a distributed Connect worker with compacted internal topics. Install the JDBC source and the S3 sink, or the S3 sink's local equivalent you can already operate. If you have no object store, use a second topic as the sink target only for the dead-letter exercise and say so. Do not substitute the demo file sink and call it the lake.
3. Submit the source connector in incrementing mode. Insert three rows. Record the offsets document and the topic contents.
4. Stop the worker process and start it again. Insert a fourth row. Show that the first three are not produced again as a full reload. If one duplicate appears at the commit boundary, show the key you would dedupe on.
5. Update a row's unit_price_cents as a superuser. Show that the topic does not gain a new event. Write the sentence that makes this safe in production.
6. Configure a sink with errors.tolerance=all and a dead-letter topic. Force one bad record. Show the dead-letter headers and show that later good records still land.
7. Draw the two cursors, source offset and sink group, and state which one Spark would not use if it were reading this topic.

Expected deliverables. Connector JSON, the offsets before and after restart, the topic dump, the dead-letter record with headers, and a short note on what Module 5 still has to define.

Evaluation rubric.

You pass when you can point at the stored line_seq, at a dead-letter header, and at the revoke, and when enrollment is absent from the connector list.

Lesson Summary

Kafka Connect is the worker runtime that moves records between an external system and a topic. Distributed workers remember source positions in a compacted offset topic and remember sink positions as consumer offsets. Converters own the bytes on the topic. Single message transforms are per-record and in-process; they are a poor place to drop or join. The JDBC source in incrementing mode is an honest extract for insert-only purchase lines with a line_seq cursor, and it is a standing defect on enrollment. Sink failures belong on a dead-letter topic when you choose to skip them. The S3 prefix is raw evidence for Module 5, partitioned by event time in UTC, not a table you query as a product. A second reader of learning.progress has its own cursor and must not be operated as if it were the Spark checkpoint.

Schema Registry and Event Contracts is next. The purchase topic is plain JSON today. The moment a field is added, that plainness needs a compatibility rule.

Further Reading

• Apache Kafka Connect — workers, distributed mode, converters, and the REST API used above.
• Connect error reporting and dead-letter queues — errors.tolerance and sink dead-letter settings from KIP-298.
• Connect single message transforms — MaskField, InsertField, ValueToKey, RegexRouter, and Filter, and the one-record limit.
• Confluent JDBC source connector — mode, incrementing.column.name, and table.include.list. This connector is not in the Apache Kafka tarball.
• Confluent S3 sink connector — flush, rotation, and TimeBasedPartitioner. Confirm class names for the build you install.
• Change Data Capture Fundamentals — why the next table does not use this poll.
• /learn/kafka — the log the offset topic and the purchase topic share.