A source connector reads data from an external system and publishes it to Kafka topics. It is a configured instance of a system-specific Kafka Connect plugin, offering an alternative to writing a custom producer. A sink connector handles the opposite direction; the two operate independently. section 1.5.2 section 5.1.2
For example, a JDBC source connector can bring customer address changes from a service database into Kafka, where a separate sink connector can deliver them to another service’s database. For reference tables, a JDBC connector can read current database state; integration with a CDC tool such as Debezium instead streams changes from source logs. Choose the approach according to whether current state or individual changes matter. section 1.5.2 section 5.1.2
A source connector’s configuration and plugin must account for the source system:
- Polling and progress: tasks retrieve data at configured intervals. A source-specific offset, such as an update timestamp or incrementing ID, identifies records not yet loaded; Kafka Connect stores source offsets in its connect-offsets topic. section 7.3.5 section 7.3.9
- Parallelism: the connector assigns portions of the source to tasks, such as separate CSV files or database tables. The configuration sets the maximum task count, but the division of work depends on the source. section 7.3.5
- Serialization: the connector translates source data into ConnectRecord; a converter then serializes it for Kafka, for example as Avro. section 7.3.3
Loading transactions for an operational data store
A JDBC source connector can load an immutable transaction table directly into Kafka. An incrementing ID marks new rows. Install the plugin, configure the database connection, table, task count, incrementing mode, and converter, then deploy the configuration through Kafka Connect’s REST API. This reduces custom development, but direct database integration can expose the database model and bypass data contracts; a separate Kafka Connect cluster is presented as a safer isolation choice. section 7.3.9 section 7.3
Listing 7.1 Configuration of the JDBC source connector
{ "name": "jdbc-source-transactions",
"config":
{"connector.class": "JdbcSource", #1
"tasks.max": 1, #2
"topic.prefix": "connect-", #3
"connection.url": "jdbc:postgresql://postgres:5432/trans_db", #4
"mode": "incrementing", #5
"incrementing.column.name":"ID", #6
"value.converter": "io.confluent.connect.avro.AvroConverter", #7
"value.converter.schema.registry.url":"http://schema-registry:8081", #8
"table.whitelist" : "public.transactions", #9
"connection.user": "demo",
"connection.password" : "demo"
}
}
The JDBC configuration makes the integration choices concrete: source table, incrementing ID, task limit, and Avro converter. section 7.3.9
