JDBC Sink Connector — PostgreSQL
Type |
Sink |
Class |
|
Target System |
Database (PostgreSQL) |
Maintainer |
Aiven |
License |
Apache License 2.0 |
Project |
|
Download |
|
This page documents version 6.12.0. Newer versions should be compatible unless there are breaking changes, but field names or default values may differ. If you notice discrepancies, please contact Axual Support. |
Description
The JDBC Sink Connector for PostgreSQL consumes records from Kafka topics and writes them into PostgreSQL tables.
It is maintained by Aiven as part of the open-source github.com/aiven/jdbc-connector-for-apache-kafka.
Features
-
Write Kafka records into PostgreSQL tables
-
Configurable write modes:
insertandupsert -
Primary key extraction from record key, value, or Kafka coordinates
-
Flexible table name mapping via configurable pattern
-
Schema-aware: requires JSON (with schema) or Avro converters
When to Use
-
You need to write Kafka records into a PostgreSQL table as part of a data pipeline.
-
You want to materialise a Kafka topic as a database table for downstream querying.
-
You need upsert semantics to keep a PostgreSQL table in sync with a Kafka topic.
When NOT to Use
-
Your Kafka records do not carry a schema — the connector requires the schema envelope to construct INSERT statements. Use a schema-aware converter upstream.
-
Your database schema changes frequently — schema evolution is not supported and requires manual reconfiguration.
Installation
The connector is available from the Maven Central.
-
Navigate to the releases page and select the version matching your Kafka Connect installation.
-
Download the JAR file.
For installation steps, see Installing Connector Plugins.
Configuration
For the complete configuration reference, see the official sink connector documentation.
| To configure a connector in Axual Self-Service, see Starting Connectors. TIP: To speed up your deployment, use the Terraform Boilerplate or the Management API Boilerplate. |
Getting Started
This section walks through setting up the JDBC Sink Connector to write records from a Kafka stream into a PostgreSQL table.
The example assumes a stream named postgresql_CUSTOMERS already contains records with a schema envelope
(e.g. produced by the JDBC Source Connector).
Prerequisites
PostgreSQL instance
You need a running PostgreSQL instance reachable from the Kafka Connect cluster, with a database and user that has write permissions.
If you do not have a PostgreSQL instance yet, see the JDBC Source Connector prerequisites for Google Cloud setup instructions.
Axual stream with schema-enabled records
The Kafka stream this connector consumes must contain records with a schema envelope.
If you are using JsonConverter, the producer must set value.converter.schemas.enable=true.
If the stream was produced by the JDBC Source Connector with value.converter.schemas.enable=true,
it is already in the correct format.
|
Steps
Step 1 — Create a connector application
-
Follow the Configure and install a connector documentation to set up a new Connector-Application.
Let’s call itmy.postgresql.sink.
The plugin name isio.aiven.connect.jdbc.JdbcSinkConnector.
If a plugin isn’t available, ask a platform operator to install plugins.
Step 2 — Configure the connector
-
Provide the following minimal configuration:
connector.classio.aiven.connect.jdbc.JdbcSinkConnectorconnection.urljdbc:postgresql://PASTE_THE_IP_ADDRESS:5432/inventoryconnection.userpostgresconnection.passwordYour database password
insert.modeupsertpk.fieldsuser_idpk.moderecord_value
Extract the primary key from the Kafka record value.table.name.formatkafka_${topic}
Writes into the tablekafkapostgresql_CUSTOMERS._topicspostgresql_CUSTOMERS
Case-sensitive.value.converterorg.apache.kafka.connect.json.JsonConvertervalue.converter.schemas.enabletrueFor advanced options, see the official sink connector documentation.
Step 3 — Start the connector
Start the connector application from Axual Self-Service.
Once running, records from postgresql_CUSTOMERS will be written to the kafka_postgresql_CUSTOMERS table in PostgreSQL.
Step 4 — Verify
Verify the data using the Google Cloud Shell:
gcloud sql connect my-axual-postgresql-connector-test --user=postgres
Then query the table:
USE inventory;
SELECT * FROM kafka_postgresql_CUSTOMERS;
Cleanup
When you are done:
-
Stop the connector application in Axual Self-Service.
-
Remove stream access for the application if no longer needed.
-
Delete your SQL instance if it was created only for testing.
Known limitations
-
The connector requires the Kafka record value to include the table schema —
value.converter.schemas.enablemust betrue. -
Schema evolution is not supported — changes to the record schema require manual reconfiguration.
-
table.name.formatvalues are case-sensitive on PostgreSQL when using quoted identifiers.
Examples
Minimal configuration
{
"name": "my-jdbc-postgresql-sink",
"config": {
"connector.class": "io.aiven.connect.jdbc.JdbcSinkConnector",
"connection.url": "jdbc:postgresql://123.123.123.123:5432/inventory",
"connection.user": "postgres",
"connection.password": "<your-database-password>",
"insert.mode": "upsert",
"pk.fields": "user_id",
"pk.mode": "record_value",
"table.name.format": "kafka_${topic}",
"topics": "postgresql_CUSTOMERS",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "true"
}
}
License
PostgreSQL JDBC Connector is licensed under the Apache License, Version 2.0.