Apache Kafka / Advanced: atomic processing and recovery design / Lesson 15

Connect, converters and change capture

Separate connector responsibilities from worker runtime, serialization and external-system guarantees.

About 4 min readContent reviewed 2026-10-28

On this page

Lesson goals

Explain source and sink connectors, tasks and converters.
Review snapshot, streaming and duplicate behavior for a CDC design.

The framework and plugin have different contracts

Kafka Connect supplies a runtime for moving data between Kafka and external systems through connector plugins. A source connector produces Kafka records from an external source; a sink connector consumes Kafka records into a destination. A connector divides work into tasks when its design supports division, and workers execute those tasks. Increasing tasks.max does not force a plugin to parallelize one indivisible source. Distributed worker mode stores coordination and progress data in internal Kafka topics, whose configuration and availability matter. Connect is not a guarantee that every installed plugin has identical security, delivery semantics or operational maturity.

A converter is not the connector itself

Connect records have a data model that converters translate to or from Kafka key and value bytes. A JSON converter's schema settings change the expected envelope, so mixing schema-enabled writer output with a reader expecting plain JSON can fail despite both sides saying JSON. Key and value converters are separate choices. Single Message Transforms can adjust individual records but should not be treated as an unrestricted multi-record transaction engine or a substitute for complex stream processing. Record the plugin version, converter settings and schema expectations together when reviewing a connector deployment or diagnosing a serialization incident.

Inspect a schema-bearing training envelope

The fixture represents a simple Connect-style JSON envelope with schema and payload. It is not a universal format for every connector and is not a complete connector configuration. A consumer reading the whole object as if id were at the top level would fail to find it. Conversely, removing the envelope without reviewing converter behavior can discard type information expected by a sink. The example deliberately uses a tiny integer field so the distinction is visible. Real records can include logical types, nested structures and nullability whose exact representation must follow the chosen converter and source plugin documentation.

json Synthetic schema-enabled JSON envelope

{"schema":{"type":"struct","fields":[{"field":"id","type":"int32","optional":false}],"optional":false},"payload":{"id":17}}

id is inside payload; converter configuration determines whether this envelope is expected.

A snapshot and a change stream need a shared boundary

Change-data-capture systems often combine an initial snapshot with later changes from a source log. Correctness depends on how the specific connector coordinates the snapshot boundary, source position and ongoing writes. Kafka alone cannot infer that coordination. If source log history expires before the connector catches up, recovery may require a new snapshot or another supported procedure. Deletes, primary-key changes and transaction ordering deserve explicit fixtures. A change event is also not necessarily a business-domain event: a row update can reflect maintenance or intermediate state rather than a meaningful parcel transition suitable for every consumer.

Keep source progress separate from Kafka coordinates

The local mapping stores a synthetic source position beside a Kafka coordinate and verifies they are different fields. A source offset belongs to the connector's source protocol; a Kafka partition offset belongs to the destination log. Treating one as the other makes incident recovery unsafe. Connector status reporting task RUNNING is not proof the destination has current business data, so compare source freshness, output progress and sink completion separately. No Connect worker, source database or connector plugin is started here. The model validates representation only, while connector guarantees remain conditional on official plugin documentation and authorized integration tests.

python Offline source/destination progress distinction

progress = {'source_position': 'lab-log:204', 'kafka': {'topic': 'lab-changes', 'partition': 0, 'offset': 17}}
assert progress['source_position'] == 'lab-log:204'
assert progress['kafka']['offset'] == 17
print(progress['source_position'], progress['kafka']['offset'])

Expected output

lab-log:204 17

The two coordinates use different protocols and cannot be substituted for one another.

Common errors and fixes

The sink rejects a JSON record with an unexpected envelope.

Why it happens: Converter schema modes differ between writer and reader.
How to fix it: Align the documented converter contract and inspect exact serialized fixtures.
Prevent it next time: Version converter settings alongside connector configuration.

Increasing tasks.max creates no additional useful tasks.

Why it happens: The plugin or source cannot divide that work further.
How to fix it: Review the connector’s task model and actual bottleneck.
Prevent it next time: Do not infer parallelism from a maximum setting alone.

A connector cannot resume from its old source position.

Why it happens: Required source-log coverage may have expired.
How to fix it: Follow the connector’s supported recovery procedure and reconcile coverage.
Prevent it next time: Monitor source-log retention against worst-case connector outage.

Check your understanding

1. Does a RUNNING task prove the dashboard has processed all current source changes?

Show worked solution

No. It describes task state, not end-to-end business freshness. The source can be behind, the task can be retrying, Kafka output can await consumption or the sink can lag. Compare source position, Kafka progress and destination completion with aligned timestamps and explicit metric definitions.

2. Why can a database row-change topic be unsuitable as a stable public business-event API?