Using Kafka Connect JDBC Source: a PostgreSQL example
Tags: apachekafka, kafkaconnect, jdbc, postgres
Published: 2021-03-25 00:00:00

---
title: Using Kafka Connect JDBC Source: a PostgreSQL example
published: true
date: 2021-03-25 00:00:00 UTC
tags: apachekafka, kafkaconnect, jdbc, postgresql
canonical_url: https://aiven.io/blog/using-kafka-connect-jdbc-source-a-postgresql-example
---
![Football Hero Image](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/ys4itlw2zn1hx4xiii92.png)

# Using Kafka Connect JDBC Source: a PostgreSQL example

If we go back in history few years, the typical data pipeline was an app creating events and pushing them to a backend database. Data was then propagated to downstream applications via dedicated ETL flows at regular intervals, usually daily.
In these modern times, [Apache Kafka](https://aiven.io/kafka?utm_source=devto&utm_medium=organic&utm_campaign=blog_art&utm_content=post) has become the default data platform. Apps write events to Kafka, which then distributes them in near-real-time to downstream sinks like databases or cloud storages. [Kafka Connect](https://aiven.io/kafka-connect?utm_source=devto&utm_medium=organic&utm_campaign=blog_art&utm_content=post), a framework to stream data into and out of Apache Kafka, represents a further optimisation that makes the ingestion and propagation of events just a matter of config files settings.

What if we're facing an old app-to-database design? How can we bring it to 2021 and include Kafka in the game? Instead of batch exporting to the database at night, we can add Kafka to the existing system. Kafka Connect lets us integrate to an existing system and make use of more up-to-date tech tools, without disrupting the original solution.

One way to do this is to use the Kafka Connect JDBC Connector. This post will walk you through an example of sourcing data from an existing table in PostgreSQL and populating a Kafka topic with only the changed rows. This is a great approach for many use cases. But when no additional query load to the source system is allowed, you could also make use of change data capture solutions based on tools like Debezium. As we'll see later, [Aiven provides Kafka Connect as a managed service](https://aiven.io/kafka-connect) for both options. You can start your connectors without the hassle of managing a dedicated cluster.

![Architecture Overview](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/houjd12zrwr4s5cfgnvc.png)

This blog post provides an example of the Kafka Connect JDBC Source based on a PostgreSQL database. A more detailed explanation of the connector is provided in our [help article](https://help.aiven.io/en/articles/3416789-jdbc-source-connector-with-postgresql)

In our example, we first create a PostgreSQL database to act as backend data storage for our imaginary application. Then we create a Kafka cluster with Kafka Connect and show how any new or modified row in PostgreSQL appears in a Kafka topic.

## Creating the PostgreSQL Source system

We'll create the whole setup using the [Aiven Command Line Interface](https://aiven.io/blog/command-line-magic-with-the-aiven-cli). Follow the instructions in the help article to install and log in. All you need is Python 3.5+ and an Internet connection.

Once you've logged in to the Aiven client, we can create a PostgreSQL database with the following `avn` command in our terminal:

```bash
avn service create pg-football    \
    -t pg                         \
    --cloud google-europe-west3   \
    -p business-4
```

This command creates a PostgreSQL database (flag `-t pg`) named `pg-football` on region `google-europe-west3`. The selected plan driving the amount of resources available and associated billing is `business-4`.

The create command returns immediately, Aiven received the request and started creating the instance. We can wait for the database to be ready with the following command:

```bash
avn service wait pg-football
```
The `wait` command can be executed against any Aiven instance, and returns only when the service is in `RUNNING` mode.

### Time to Scout Football Players

Now let's create our playground: we are a football scouting agency, checking players all over the world and our app pushes the relevant data to a PostgreSQL table. Let's login to PostgreSQL from the terminal:

```bash
avn service cli pg-football
```

Or agency doesn't do a great job at scouting, all we are able to capture is the player's `name`, `nationality` and a flag `is_retired` showing the activity status.
We create a simple `football_players` table containing the above information together with two control columns:
* `created_at` keeping the record's creation time
* `modified_at` for the row's last modification time

These two columns will later be used from the Kafka Connect connector to select the recently changed rows.
Now it's time to create the table from the PostgreSQL client:

```sql
CREATE TABLE football_players (
  name VARCHAR ( 50 ) PRIMARY KEY,
  nationality VARCHAR ( 255 ) NOT NULL,
  is_retired BOOLEAN DEFAULT false,
  created_at TIMESTAMP NOT NULL DEFAULT NOW(),
  modified_at TIMESTAMP
  )
;
```

The `created_at` field will work as expected immediately, with the `DEFAULT NOW()` definition.
The `modified_at` on the other side, requires a bit more tuning to be usable. We'll need to create a trigger that inserts the current timestamp in case of updates. The following SQL can be executed from the PostgreSQL client:

```sql
CREATE OR REPLACE FUNCTION change_modified_at()
  RETURNS TRIGGER
  LANGUAGE PLPGSQL
  AS
$$
BEGIN
	NEW.modified_at := NOW();
	RETURN NEW;
END;
$$
;
CREATE TRIGGER modified_at_updates
  BEFORE UPDATE
  ON football_players
  FOR EACH ROW
  EXECUTE PROCEDURE change_modified_at();
```

The first statement creates the `change_modified_at` function that will later be used by the `modified_at_updates` trigger.

### Football Scouting App at Work

We can now simulate our football scouting app behaviour by manually inserting three rows in the `football_players` table from the PostgreSQL client with

```sql
insert into football_players (name, nationality, is_retired) values ('Andrea Pirlo','Italian', true);
insert into football_players (name, nationality, is_retired) values ('Cristiano Ronaldo','Portuguese', false);
insert into football_players (name, nationality, is_retired) values ('Megan Rapinoe','American', true);
```

We can verify that the `created_at` column is successfully populated in PostgreSQL with

```sql
select * from football_players;
```

Which will output

```sql
       name        | nationality | is_retired |         created_at         | modified_at
-------------------+-------------+------------+----------------------------+-------------
 Andrea Pirlo      | Italian     | t          | 2021-03-11 10:35:52.04076  |
 Cristiano Ronaldo | Portuguese  | f          | 2021-03-11 10:35:52.060104 |
 Megan Rapinoe     | American    | t          | 2021-03-11 10:35:52.673554 |
(3 rows)
```

Perfect, the app is working when inserting new rows. If only we could have an update to an existing row...

![Breaking News Pirlo](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/t71f4llqi5lktne8ctix.png)

Well, this was somehow expected, Juventus FC went out of Champions League and needed new energy in the midfield. We can update the relevant row with

```sql
update football_players set is_retired=false where name='Andrea Pirlo';
```

We can check that the `modified_at` is correctly working by issuing the same `select * from football_players;` statement in the PostgreSQL client and checking the following output

```sql
      name        | nationality | is_retired |         created_at         |        modified_at         
------------------+-------------+------------+----------------------------+----------------------------
Cristiano Ronaldo | Portuguese  | f          | 2021-03-11 10:35:52.060104 |
Megan Rapinoe     | American    | t          | 2021-03-11 10:35:52.673554 |
Andrea Pirlo      | Italian     | f          | 2021-03-11 10:35:52.04076  | 2021-03-11 10:39:49.198286
(3 rows)
```

Ok, we recreated the original setup: our football scouting app is correctly storing data in the `football_players` table. In the old days the extraction of that data was demanded to an ETL flow running overnight and pushing it to the downstream applications. Now, as per our original aim, we want to include Apache Kafka in the game, so... let's do it!

## Creating a Kafka environment

As stated initially, our goal is to base our data pipeline on Apache Kafka without having to change the existing setup. We don't have a Kafka environment available right now, but we can easily create one using Aiven's CLI from the terminal with the following `avn` command

```bash
avn service create kafka-football         \
  -t kafka                                \
  --cloud google-europe-west3             \
  -p business-4                           \
  -c kafka.auto_create_topics_enable=true \
  -c kafka_connect=true
```

The command creates an Apache Kafka instance (`-t kafka`) in `google-europe-west3` with the `business-4` plan.
Additionally it enables the topic auto-creation (`-c kafka.auto_create_topics_enable=true`) so our applications can create topics on the fly without forcing us to create them beforehand.
Finally, it enables Kafka Connect (`-c kafka_connect=true`) on the same Kafka instance. We can use the `avn wait` command mentioned above to pause until the Kafka cluster is in `RUNNING` state.

Note that on Kafka instances part of the `startup` plans, you'll be forced to create a standalone Kafka Connect instance. For production systems, we recommend using standalone Kafka Connect for the separation of concerns principle.

### Connecting the dots

The basic building blocks are ready: our source system represented by the `pg-football` PostgreSQL database with the `football_players` table and the `kafka-football` Apache Kafka instances are running. It's now time to connect the two: creating a new event in Kafka every time an insert or modified row appears in PostgreSQL. That can be achieved by creating a Kafka Connect JDBC source connector.

## Create a JSON configuration file

Start by creating a JSON configuration file like the following:

```json
{
    "name": "pg-timestamp-source",
    "connector.class": "io.aiven.connect.jdbc.JdbcSourceConnector",
    "connection.url": "jdbc:postgresql://<HOSTNAME>:<PORT>/<DATABASE>?sslmode=require",
    "connection.user": "<PG_USER>",
    "connection.password": "<PG_PASSWORD>",
    "table.whitelist": "football_players",
    "mode": "timestamp",
    "timestamp.column.name":"modified_at,created_at",
    "poll.interval.ms": "2000",
    "topic.prefix": "pg_source_"
}
```

Where the important parameters are:
* `name`: the name of the Kafka Connect connector, in our case `pg-timestamp-source`
* `connection.url`: the connection URL pointing to the PostgreSQL database, in the form of `jdbc:postgresql://<HOSTNAME>:<PORT>/<DATABASE>?<ADDITIONAL_PARAMETERS>`, we can create it with the `dbname`, `host`, `port` output of the following `avn` command

```bash
avn service get pg-football --format '{service_uri_params}'
```

* `connection.user` and `connection.user`: PostgreSQL credentials, the default `avnadmin` credentials are available as `user` and `password` output of the `avn` command above
* `table.whitelist`: list of tables to source from PostgreSQL, in our case is  `football_players`
* `mode`: Kafka Connect JDBC mode. Three modes are available: `bulk`, `incrementing`, `timestamp`. For this post we'll use the `timestamp` one. For a more detailed description of modes, please refer to the [help article](https://help.aiven.io/en/articles/3416789-jdbc-source-connector-with-postgresql)
* `timestamp.column.name`: list of timestamp column names: The value for this setting should be `modified_at,created_at` since `modified_at` will contain the most recent update timestamp, and in case of null value, we can rely on the `created_at` column.
* `poll.interval.ms`: time between database polls
* `topic.prefix`: prefix for topic, the full topic name will be a concatenation of `topic.prefix` and the PostgreSQL table name.

### Start the JDBC connector

After storing the above JSON in a file named `kafka_jdbc_config.json`, we can now start the Kafka Connect JDBC connector in our terminal with the following command:

```bash
avn service connector create kafka-football @kafka_jdbc_config.json
```

We can verify the status of the Kafka Connect connector with the following `avn` command:

```
avn service connector status kafka-football pg-timestamp-source
```
Note that the last parameter `pg-timestamp-source` in the `avn` command above refers to the Kafka Connect connector name defined in the `name` setting of the `kafka_jdbc_config.json` configuration file. If all settings are correct, the above command will show our healthy connector being in `RUNNING` mode

```
{
    "status": {
        "state": "RUNNING",
        "tasks": [
            {
                "id": 0,
                "state": "RUNNING",
                "trace": ""
            }
        ]
    }
}
```

### Check the data in Kafka with Kafkacat

The data should now have landed in Apache Kafka. How can we check it?
We can use [Kafkacat](https://help.aiven.io/en/articles/2607674-using-kafkacat) a nice command line utility.

Once Kafkacat is installed (see our [help article](https://help.aiven.io/en/articles/2607674-using-kafkacat) for detailed instructions), we'll need to set up the connection to our Kafka environment.

Aiven by default enables SSL certificate based authentication. The certificates are available from the Aiven console for manual download. In Aiven CLI you can avoid the clicking with the following `avn` command in our terminal:

```bash
mkdir -p kafkacerts
avn service user-creds-download kafka-football \
    -d kafkacerts \
    --username avnadmin
```

These commands create a `kafkacerts` folder (if not existing already) and download in it the `ca.pem`, `service.cert` and `service.key` SSL certificates required to connect.

The last missing piece of information that Kafkacat needs is where to find our Kafka instance in terms of `hostname` and `port`. This information can be displayed in our terminal with the following `avn` command

```bash
avn service get kafka-football --format '{service_uri}'
```

Once we collected the required info we can create a `kafkacat.config` file with the following entries

```
bootstrap.servers=<KAFKA_SERVICE_URI>
security.protocol=ssl
ssl.key.location=kafkacerts/service.key
ssl.certificate.location=kafkacerts/service.cert
ssl.ca.location=kafkacerts/ca.pem
```

Remember to substitute the `<KAFKA_SERVICE_URI>` with the output of the `avn service get` command mentioned above.

Now we are ready to read the topic from Kafka by pasting the following command in our terminal:

```
kafkacat -F kafkacat.config -C -t pg_source_football_players
```

Note that we are using Kafkacat in consumer mode (flag `-C`) reading from the topic `pg_source_football_players` which is the concatenation of the `topic.prefix` setting in Kafka Connect and the name of our `football_players` PostgreSQL table.

As expected, since the connector is working, Kafkacat will output the three messages present in the Kafka topic matching the three rows in the `football_players` PostgreSQL table

```json
{"name":"Cristiano Ronaldo","nationality":"Portuguese","is_retired":false,"created_at":1615458952060,"modified_at":null}
{"name":"Megan Rapinoe","nationality":"American","is_retired":true,"created_at":1615458952673,"modified_at":null}
{"name":"Andrea Pirlo","nationality":"Italian","is_retired":false,"created_at":1615458952040,"modified_at":1615459189198}
% Reached end of topic pg_source_football_players [0] at offset 3
```
## Updating the listings

Now, let's see if our football scouts around the world can fetch some news for us

![Breaking News Two Updates](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/kqaysqgpby6s49vgj6ww.png)

Wow, we found a new talent named `Enzo Gorlami` and `Cristiano Rolando` officially retired today from professional football (please be aware this post is not reflecting football reality). Let's push the two news to PostgreSQL:

```sql
insert into football_players (name, nationality, is_retired) values ('Enzo Gorlami','Italian', false);
update football_players set is_retired=true where name='Cristiano Ronaldo';
```

We can verify that the data is correctly stored in the database:
```sql
defaultdb=> select * from football_players;
       name        | nationality | is_retired |         created_at         |        modified_at         
-------------------+-------------+------------+----------------------------+----------------------------
 Megan Rapinoe     | American    | t          | 2021-03-11 10:35:52.673554 |
 Andrea Pirlo      | Italian     | f          | 2021-03-11 10:35:52.04076  | 2021-03-11 10:39:49.198286
 Enzo Gorlami      | Italian     | f          | 2021-03-11 11:09:49.411885 |
 Cristiano Ronaldo | Portuguese  | t          | 2021-03-11 10:35:52.060104 | 2021-03-11 11:11:36.790781
(4 rows)
```

And in Kafkacat we receive the following two updates:

```json
{"name":"Enzo Gorlami","nationality":"Italian","is_retired":false,"created_at":1615460989411,"modified_at":null}
% Reached end of topic pg_source_football_players [0] at offset 4
{"name":"Cristiano Ronaldo","nationality":"Portuguese","is_retired":true,"created_at":1615458952060,"modified_at":1615461096790}
% Reached end of topic pg_source_football_players [0] at offset 5
```
## Further reading
If you want to know more about Aiven, Kafka, Kafka Connect or PostgreSQL, check the references below:

* [Kafka Connect JDBC Source PostgreSQL](https://help.aiven.io/en/articles/3416789-jdbc-source-connector-with-postgresql)
* [Intro to PostgreSQL](https://aiven.io/blog/an-introduction-to-postgresql)
* List of [Aiven supported Kafka Connect Connectors](https://help.aiven.io/en/articles/1231452-kafka-connect-connectors)
* [Kafka Connect Preview](https://aiven.io/blog/kafka-connect-preview)
* [Aiven Command Line Interface](https://help.aiven.io/en/articles/3074410-aiven-command-line-interface-tips-and-tricks)

## Wrapping up
This blog post showed how to easily integrate PostgreSQL and Kafka with a fully managed, config-file-driven Kafka Connect JDBC connector. We used a timestamp-based approach to retrieve the changed rows since the previous poll and push them to a Kafka topic increasing the query load to the source database.

An alternative method is represented by Change Data Capture solutions like Debezium, which, in case of PostgreSQL, reads changes directly from WAL files avoiding any additional query load on the source database. A guide on how to setup CDC for Aiven PostgreSQL is provided in this [help article](https://help.aiven.io/en/articles/1790791-setting-up-debezium-with-aiven-for-postgresql).