An inventory database records that a store has 27 units of a product available. A reporting database needs that information, and an archive needs to retain the inventory updates for later analysis. Kafka can carry the data to both destinations, but something must read the inventory system and something must write to each destination.

Kafka Connect divides those jobs into source connectors and sink connectors. Their direction is easy to remember. A correct pipeline depends on how each side captures changes, interprets records, and saves progress.

In this chapter, we'll follow an inventory update through both sides and compare the decisions each connector must make.

A source connector moves data from another system into Kafka. A sink connector reads data from Kafka and writes it to another system. Source and sink describe a role in a particular data flow, not a permanent property of a database or service.

For example, PostgreSQL can be a source when Connect reads inventory changes from it. A different PostgreSQL database can be a sink when Connect writes those records into reporting tables. Supporting PostgreSQL in one direction does not imply that the same plugin supports the other direction.

Our inventory pipeline has one source integration and two independent sink integrations:

The reporting sink and archive sink use separate consumer groups so each can read all the inventory records. They do not divide the records between the two destinations.

The arrows show logical data movement. The actual transfer runs in tasks, the units of work that Connect workers execute. Source tasks supply records to Connect's producer path; sink tasks receive records from its consumer path and write them externally.

A source does not need to know which sinks will read its output. A sink can also consume records that an ordinary application producer writes. Their shared contract is the Kafka topic and the records it contains.

A source must decide what to read, how to represent it as records, and how to identify its reading position. Those decisions depend on the external system and the plugin.

For a database, a connector might periodically query rows. Another might capture committed changes from the database's transaction log. A file source could read appended lines, while an API source might request new items using a service-provided cursor, a token identifying where to continue reading.

These methods can produce different information even when they read the same underlying data.

Current State and Change History

Suppose the stock count changes from 30 to 28, then from 28 to 27 between two database queries. A query that reads the current row may observe only 27. A log-based change capture connector can observe both committed updates if the database exposes them and the connector captures the relevant changes.

Reading 27 may be sufficient for a current-stock report. It is insufficient for an audit that needs every stock movement. Publishing query results to Kafka does not recover intermediate changes that the query never observed.

The query's selection rule matters too. A source that selects only rows with a newly increasing ID can find new inventory items, but it will not detect quantity changes to an existing row. Timestamp-based selection can detect updates only if the required columns and query behavior correctly identify them.

Initial data needs a decision as well. If the reporting database starts empty, capturing only future changes leaves unchanged inventory items missing. Some connectors support an initial snapshot, a read of existing data, followed by ongoing change capture. The handoff between those phases is connector-specific.

Building the Record

For our example, assume a source captures committed inventory changes and the pipeline publishes the current stock fields after any configured transformation. It writes to inventory.stock with key sku-1042 and this value:

Here, sku identifies a product, and we assume one aggregate inventory row per product. The inventory application maintains an increasing version for that row. The JSON illustrates the logical value; it is not a claim about any connector's default wire format.

This record means “the available quantity is 27.” It does not mean “add 27 units.” That distinction controls what a destination must do when it reads the record, especially if it receives it again.

A database primary key also does not automatically become a Kafka record key for every source plugin. The team must verify or configure that mapping. Our example explicitly uses sku-1042 as the Kafka key so a destination can identify which inventory row to update.

A sink starts with Kafka records and turns them into operations its destination understands. The reporting sink might write SQL rows, while the archive sink might collect records into files and upload them to object storage.

Both receive the stock value of 27, but they preserve different views of it.

The reporting database needs one current row per product. Assume its sink supports an upsert, which inserts a row if it does not exist or updates the existing row that a key identifies. The sink maps sku-1042 to the destination's primary key and sets availableUnits to 27 and version to 86.

The archive needs a sequence of observations over time. Its sink can preserve this record alongside earlier records instead of replacing the earlier stock value. File creation, batching, and duplicate handling depend on the archive connector.

Simply selecting a destination does not choose the right write behavior. An insert-only reporting sink could fail on a duplicate primary key when the next update arrives. If the table has no suitable uniqueness constraint, it could accumulate multiple rows where the report expects one.

A sink may also buffer records before writing a batch. Receiving a record inside the task is therefore different from completing its destination write. Progress must reflect writes that are safe to consider complete, according to the connector's delivery contract.

Source configuration primarily describes what to capture and where to publish it. Sink configuration describes which topics to consume and how to apply their records. Details such as table names, key mappings, write modes, and file layout belong to the selected plugin.

Two connectors can each support Kafka and still disagree about the data they exchange. They need compatible serialization, record structure, and meaning.

A converter translates between Connect's internal data representation and Kafka's stored bytes. On the source path, conversion to bytes happens before publication. On the sink path, conversion from bytes happens before the sink task receives the record.

This diagram shows the conversion boundary for the record value, omitting optional transformations. Keys have their own converter configuration.

The sink converter must understand the encoding the producer uses. But successful decoding is only the first step. A sink may require an explicit schema describing field names and types, or it may accept only certain record shapes. Two integrations that both claim to use JSON can still disagree about whether schema information accompanies the value.

Our reporting sink expects stock fields it can map to columns. A change capture connector might instead produce an envelope, a structure containing the operation type, old row, new row, and source metadata. The sink needs either native support for that envelope or a deliberate transformation into the expected shape.

Deletes

Suppose the inventory system removes a discontinued product. The source must first detect that removal. A query that returns only existing rows does not automatically emit a record for a row that disappeared.

The pipeline must then represent the deletion in a form the sink understands. One possible contract uses a tombstone, a record with a non-null key and a null value. For our product, that would be key sku-1042 with a null value, rather than a JSON object whose fields happen to contain nulls.

Kafka uses tombstones in log-compacted topics to represent deletion of a key. That does not issue a delete against an external database. Depending on the plugin and configuration, a sink might delete the matching row, ignore the record, or fail.

For example, Confluent's JDBC sink supports tombstone-driven row deletion when you enable deletion and the primary key comes from the record key. That is a feature with specific requirements in that connector, not a rule shared by all sinks.

For the inventory report to stay correct, the entire path must agree: capture the deletion, encode it deliberately, and configure the sink to apply it.

Sources and sinks both save progress, but their checkpoints identify positions in different systems.

A source supplies a source partition, an identifier for an external input stream, and a source offset, a connector-defined position within it. These might identify a database change stream and a log position. They are separate from Kafka topic partitions and offsets; there is no required one-to-one mapping between them.

A sink normally tracks consumed Kafka offsets through its consumer group. Suppose our stock update lands in inventory.stock, partition 0, at offset 120. After successfully applying it and all preceding records, the reporting sink can commit 121 as its next restart position for that partition.

The sequence below shows the two failure windows, source first and then sink. It assumes at-least-once operation, retained input data, and recovery from the saved checkpoint. C86 is a symbolic external change position, not a real database offset format.

The source failure can create two Kafka records representing the same external change. The sink failure can apply one Kafka record twice without adding another record to the topic. Deduplicating only by Kafka topic, partition, and offset can recognize the second case, but it cannot recognize the first because the duplicate publication has a different Kafka offset.

For our current-stock table, repeating “set quantity to 27” for the same key can be harmless. Repeating “increase quantity by 27” would change the result. If older updates can arrive after newer ones, the destination also needs supported version checks or ordering controls to prevent version 86 from overwriting version 87. Including a version field is useful only when the write logic checks it.

Task reassignment does not remove these failure windows. A replacement task still resumes from saved progress. Recovery also depends on the external source retaining the required input and Kafka retaining the records the sink needs.

These examples do not describe every connector's guarantee. Connect supports exactly-once source publication for compatible connectors in distributed mode when the operator enables and configures it appropriately. Sink guarantees depend on how the plugin coordinates Kafka progress with destination writes. Neither connector's direction alone establishes an exactly-once guarantee across the entire pipeline.

The practical distinction is what each side must understand about its input and output:

The inventory pipeline needs both sides to be correct. A reporting sink cannot reconstruct a stock change that the source never captured. A source that captures every change cannot prevent a sink from using the wrong primary key or treating a current quantity as an increment.

Their progress can also diverge. If the reporting database is unavailable, the source may continue publishing while the reporting sink falls behind. The archive can continue independently. Kafka provides the retained records that make catching up possible, subject to storage capacity and cleanup policies.

Evaluate the source against what the business needs to observe, and the sink against the state or history the destination must preserve. Then check the record contract between them with inserts, updates, deletes, and repeated delivery.

Source connectors capture external data and publish Kafka records. Sink connectors consume those records and apply them to external systems. Sources track where they have read externally; sinks normally track their progress through Kafka partitions.

A working pipeline needs agreement on more than direction. Capture behavior, keys, schemas, update and delete semantics, and recovery behavior determine whether the destination contains the intended data.