A JDBC source plugin lets Kafka Connect read relational database data through JDBC and publish it to Kafka without custom producer code. The source connector maps tables to Kafka topics, can distribute tables across tasks, and can adapt to added or removed tables. JDBC drivers determine which database systems are supported. section 7.3.8 section 1.5.2
Choose a polling mode according to the table’s change and durability characteristics:
- Bulk periodically loads the entire table, but can create duplicates and is unsuitable for large tables. section 7.3.8
- Incrementing tracks new rows through an increasing column, typically a sequence or primary key; it detects inserts but not updates. section 7.3.8 section 7.3.9
- Timestamp tracks rows through a last-modified timestamp, while incrementing plus timestamp combines both tracking signals. section 7.3.8
- Query uses custom SQL and can combine tracking with joins for enrichment or consolidation across related tables. section 7.3.8
Customer 360 ODS transactions
An ODS can install Kafka Connect, deploy the JDBC source plugin, and submit a connector configuration through the Connect REST API. For immutable transactions, incrementing mode uses the ID column to load new rows efficiently into the transactions topic. A dedicated Kafka Connect cluster isolates integration work from the application database and avoids exposing the full database model through custom scripts. section 7.3.9
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"
}
}
Shows a concrete connector configuration with a JDBC URL, incrementing mode, ID tracking, a transactions table allowlist, Avro conversion, and Schema Registry. section 7.3.9
