Importing Data from a Database with Kafka Connect

1. Create a JDBC connector to load data from the PostgreSQL database.

curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
  "name": "jdbc_source_postgres",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "connection.url": "jdbc:postgresql://10.0.1.102:5432/inventory",
    "connection.user": "kafka",
    "connection.password": "Kafka!",
    "topic.prefix": "postgres-",
    "mode":"timestamp",
    "timestamp.column.name": "update_ts"
  }
}'

1. List the topic. We should see a new topic called postgres-purchases.

kafka-topics --bootstrap-server localhost:9092 --list

1. Start a console consumer to read from the postgres-purchases topic:

kafka-console-consumer --bootstrap-server localhost:9092 --topic postgres-purchases --from-beginning

Note: We should see a series of records loaded from the database. Leave the consumer running so we can observe what happens when inserting a new record into the PostgreSQL table.

1. Log in to the PostgreSQL database server.

2. Change to the postgres user and connect to the inventory database:

sudo -i -u postgres

psql

\c inventory;

1. Insert a new record into the purchases table:

insert into purchases (product, quantity, purchase_time, update_ts) VALUES ('plums', 8, current_timestamp, current_timestamp);

Note: If we leave the console consumer running, we should see a new record appear in the postgres-purchases topic containing the data we just inserted.