Kafka Connect

A converter serializes/deserializes Connect data to/from bytes on the Kafka topic. key.converter handles the record key; value.converter handles the record value. They are configured independently and can differ.

errors.tolerance decides what Connect does when a record fails processing. none (the default) stops the connector task on the first error; all skips the bad record and keeps going.

Kafka Connect is a framework for streaming data between Kafka and external systems (databases, files, S3) using ready-made connectors, so you configure plugins instead of writing custom producer/consumer code.

Source connectors store their own progress (where they read from the external system) in a special Kafka topic called the offset storage topic. Sink connectors just use a normal Kafka consumer group, so Kafka tracks their offsets like any consumer.

Send a POST to /connectors with a JSON body containing a top-level "name" and a "config" object (which must repeat connector.class and the connector's settings). Connect returns 201 Created with the connector info.

Run multiple Connect worker processes with the same group.id. You scale out by starting more workers. The cluster automatically spreads connectors and their tasks evenly across all available workers via a rebalance.

Put the broker security settings (security.protocol, SASL/SSL configs) in the worker properties file. By default they apply to all of Connect's internal clients. Prefix with producer., consumer., or admin. to target a specific client type.

An SMT is a small function that modifies each record as it flows through a Kafka Connect connector. You list transforms by alias in the transforms config, then configure each alias with transforms.<alias>.type and its properties. They run in the order listed.

A Source connector pulls data from an external system INTO Kafka topics. A Sink connector reads data FROM Kafka topics and writes it OUT to an external system. Source = into Kafka, Sink = out of Kafka.

Each Debezium event has a value with op (c/u/d/r), before (old row), after (new row), and source metadata. A delete sends an event with op='d', before populated, after null — followed by a separate tombstone message (same key, null value) so log-compacted topics can drop the key.

JsonConverter serializes Connect data as JSON. With schemas.enable=true it wraps the payload as {"schema":...,"payload":...} so the schema travels inline. With false it emits just the raw JSON value with no schema.

Set errors.tolerance=all plus errors.deadletterqueue.topic.name=<topic>. Failed records are written to that topic. errors.deadletterqueue.context.headers.enable=true adds Kafka headers describing why and where the record failed.

Standalone runs a single worker and stores offsets in a local file — simple, no fault tolerance, good for dev. Distributed runs multiple coordinated workers that store config, offsets, and status in Kafka topics, giving scalability and fault tolerance.

A source task hands records to the framework, which produces them to Kafka. Periodically (every offset.flush.interval.ms, default 60000 ms) the framework flushes the producer, then writes the source offsets of acknowledged records to offset.storage.topic. Offsets are only committed after the data is safely in Kafka.

/config returns the desired configuration; /tasks returns the per-task configs; /status returns runtime health: the connector's RUNNING/FAILED/PAUSED state plus each task's state, assigned worker, and any failure trace. Status is live; config is static intent.

The old 'eager' protocol was stop-the-world: on any membership change every worker revoked ALL its connectors and tasks, then everything was reassigned. So one worker joining or a brief blip paused the entire cluster's data flow.

A ConfigProvider resolves ${provider:[path:]key} placeholders at runtime so secrets aren't stored literally in connector configs. You register a provider in the worker config (e.g. config.providers=file with FileConfigProvider) and reference values like ${file:/secrets.properties:db.password}.

It's a worker-level setting that controls whether individual connectors can override the worker's Kafka client configs using producer.override./consumer.override./admin.override. prefixes. Defaults to None (no overrides allowed); common values are All or Principal.

InsertField adds a field; ReplaceField renames/drops fields; MaskField hides values; RegexRouter and TimestampRouter rewrite the target topic name; ExtractField pulls one field up to be the whole key/value; Cast changes field types; Flatten collapses nested structs into dotted top-level fields.

A Connector is a single coordinator instance that splits work into Tasks, the things that actually move data. taskConfigs(maxTasks) returns the config for each task; tasks.max caps how many tasks Connect will run for that connector.

On first start, Debezium reads the current contents of the captured tables (the snapshot) and emits each row as an op='r' event, recording the log position (binlog offset / LSN) at snapshot start. It then begins streaming the transaction log from that recorded position, so every change after the snapshot is captured exactly once.

AvroConverter and ProtobufConverter store schemas in Schema Registry and put only a schema ID + binary payload on the topic; JsonConverter embeds or omits the schema inline with no registry. The registry-backed ones give compact messages and enforced compatibility.

config.storage.topic holds connector/task configs (single partition, compacted). offset.storage.topic holds source-connector read positions (many partitions, compacted). status.storage.topic holds connector/task/worker status (many partitions, compacted). All need high replication.

KIP-415 stops revoking everything. Only tasks that must move are revoked; the rest keep running, so there's no cluster-wide pause. It uses two rebalance rounds (revoke, then assign). The connect.protocol setting selects eager, compatible, or sessioned.

A predicate is a per-record boolean test you attach to an SMT so the SMT only runs when the predicate is true. You define predicates under predicates, reference one via transforms.<alias>.predicate, and flip it with transforms.<alias>.negate=true. Built-ins include TopicNameMatches (regex on topic) and HasHeaderKey.

SourceTask.poll() returns new SourceRecords for Connect to produce into Kafka; Connect tracks the source offsets you embed. SinkTask.put() receives batches of SinkRecords to write downstream; flush()/preCommit() is where the sink ensures records are durable before Kafka offsets are committed.

Debezium tracks the source table structure so it can correctly decode log entries that only carry column positions/values. MySQL persists captured DDL to an internal schema history topic, and can also publish DDL to a separate schema change topic for consumers. When columns are added/changed, the event schema evolves accordingly.

Connect represents data internally with a Schema (the type/shape) and a Struct (the values), independent of any wire format. Converters translate between this Schema/Struct model and the raw bytes on the topic.

StringConverter treats data as plain text (UTF-8 by default). ByteArrayConverter passes raw bytes through untouched (no schema). header.converter serializes record headers separately, defaulting to SimpleHeaderConverter.