Whitelists and Custom Query JDBC Examples

This section provides common usage scenarios using whitelists and custom queries.

Using Whitelists

Use a whitelist to limit changes to a subset of tables in a MySQL database, using id and modifiedcolumns that are standard on all whitelisted tables to detect rows that have been modified. This mode is the most robust because it can combine the unique, immutable row IDs with modification timestamps to guarantee modifications are not missed even if the process dies in the middle of an incremental update query.

The following is an example of a whitelist.

NOTE Before running this example, you need to create the stream /kafka-connect
name=mysql-whitelist-timestamp-source 
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector tasks.max=10  
connection.url=jdbc:mysql://mysql.example.com:3306/my_database?user=alice&password=secret 
table.whitelist=users,products,transactions  
mode=timestamp+incrementing 
timestamp.column.name=modified 
incrementing.column.name=id  
topic.prefix=/kafka-connect:mysql-

Using Custom Queries

Use a custom query instead of loading tables to join data from multiple tables. As long as the query does not include its own filtering, you can still use the built-in modes for incremental queries (in this case, using a timestamp column).
NOTE This limits you to a single output per connector and because there is no table name, the topic “prefix” is actually the full topic name in this case.

The following is an example of a custom query.

NOTE Before running this example, you need to create the stream /kafka-connect
name=mysql-whitelist-timestamp-source 
        connector.class=io.confluent.connect.jdbc.JdbcSourceConnector 
        tasks.max=10  
        connection.url=jdbc:postgresql://postgres.example.com/test_db?user=bob&password=secret&ssl=true 
        query=SELECT users.id, 
        users.name, 
        transactions.timestamp, 
        transactions.user_id, 
        transactions.payment FROM users JOIN transactions ON (users.id = transactions.user_id) 
        mode=timestamp 
        timestamp.column.name=timestamp  
        topic.prefix=/kafka-connect:mysql-joined-data