<a id="postgres-source-connector"></a>

#### IMPORTANT
Confluent only provides support for specific versions of Debezium-built connectors on Confluent Platform. For complete details on the latest supported Debezium 3.1.2 version, see the
official [Debezium PostgreSQL Source Connector](https://debezium.io/documentation/reference/3.1/connectors/postgresql.html) documentation.

# Debezium PostgreSQL Source Connector for Confluent Platform

The [Debezium PostgreSQL Connector](https://debezium.io/docs/connectors/postgresql/) is a source connector that
can obtain a snapshot of the existing data in a PostgreSQL database and then
monitor and record all subsequent row-level changes to that data. All of the
events for each table are recorded in a separate Apache Kafka® topic, where they can
be easily consumed by applications and services.

#### IMPORTANT
<!-- WARNING: THIS IS A SHARED FILE AND THE SOURCE IS LOCATED IN DOCS-COMMON. DO NOT ADD TO ANY OTHER REPO. -->

Effective July 6, 2025, only self-managed connector versions that meet or exceed the minimum version listed on the
[Supported Connector Versions](https://docs.confluent.io/platform/7.8/connect/supported-connector-version.html#supported-connector-versions-till-cp-7-8)
page receive support from Confluent. Older, unsupported connector versions have been removed from Confluent Marketplace and
are no longer available for download.

- Confluent supports these Debezium PostgreSQL Source connector versions:
  - versions 2.5.0 and later for Confluent Platform 7.9
  - versions 2.5.0 and later for Confluent Platform 8.0
  - versions 3.1.2 and later for Confluent Platform 8.1
- Confluent supports using this connector with PostgreSQL 12, 13, 14, 15, 16, 17, and 18.
- Confluent supports PostgreSQL 18 only with Debezium connector version 3.1.2 and later.
- Databases hosted by a service such as [Heroku Postgres](https://www.heroku.com/postgres) can’t be monitored with Debezium, since
  you may be unable to install the [logical decoding](https://www.postgresql.org/docs/9.6/logicaldecoding-explanation.html)
  plugin.

## Features

The Debezium PostgreSQL Source connector includes the following features:

- [At least once delivery](#debezium-postgresql-source-at-least-once-delivery)
- [Supports one task](#debezium-postgresql-source-supports-one-task)
- [Automatic topic creation](#debezium-postgresql-source-auto-topic-creation)
- [Client-side encryption](#debezium-postgresql-source-csfle-sm)

<a id="debezium-postgresql-source-at-least-once-delivery"></a>

### At least once delivery

The connector guarantees that records are delivered at least once to the Kafka
topic. If a fault occurs (for example, if there are network connectivity
issues), or the connector restarts, you may see some duplicate records in the
Kafka topic.

<a id="debezium-postgresql-source-supports-one-task"></a>

### Supports one task

The Debezium PostgreSQL Source connector supports running only one task.

<a id="debezium-postgresql-source-auto-topic-creation"></a>

### Automatic topic creation

The connector will create the internal database history Kafka topic if it doesn’t
exist.

<a id="debezium-postgresql-source-csfle-sm"></a>

### Client-side encryption

This connector supports Client-Side Field Level Encryption (CSFLE) and Client-Side Payload Encryption (CSPE). For more information, see [Manage Client-Side Encryption](https://docs.confluent.io/platform/current/connect/manage-csfle.html).

## Schema definition considerations

For naming and versioning consistency, Debezium schemas are defined in a central
point, unlike in earlier versions. If you use Schema Registry, you may experience schema
compatibility issues. The current workaround is to set the schema compatibility
to `NONE` if you want to upgrade from version 1.x. For more details about
Debezium schema naming and versioning, see the [Debezium 2.0.0 release](https://debezium.io/blog/2022/10/17/debezium-2-0-final-released/#all_schemas_named_and_versioned)
documentation.

## Install the Postgres connector

You can install this connector by using the [instructions](https://docs.confluent.io/kafka-connectors/self-managed/confluent-hub/client.html) or you can
manually download the ZIP file.

```bash
confluent connect plugin install debezium/debezium-connector-postgresql:latest
```

You can install a specific version by replacing `latest` with a version number. For example:

```bash
confluent connect plugin install debezium/debezium-connector-postgresql:<version-number>
```

The Debezium PostgreSQL Source connector has specific ACL requirements. See the
[ACL requirements for Debezium Source
connectors](/cloud/current/connectors/service-account.html#additional-acl-entries) to ensure you
meet the specified requirements.

### Install the connector manually

[Download and extract the ZIP file](https://www.confluent.io/hub/debezium/debezium-connector-postgresql) for
your connector and then follow the manual connector installation
[instructions](/kafka-connectors/self-managed/userguide.html#connect-installing-plugins).

## License

The Debezium PostgreSQL connector is an open source connector and does not require a Confluent Enterprise License.

## Linux on IBM Z (s390x) support

Starting with Confluent Platform 8.2, this connector supports Linux on IBM Z (s390x).
The connector supports the same capability available on x86_64 unless
otherwise noted. For more information, see [Linux on IBM Z (s390x) support](https://docs.confluent.io/platform/current/installation/versions-interoperability.html#linux-on-ibm-z-s390x-support).

## Configuration properties

For a complete list of configuration properties for this connector, see [Configuration Reference for Debezium PostgreSQL Source Connector for Confluent Platform](postgres_source_connector_config.md#postgres-source-connector-config).

## Setting up PostgreSQL

Before using the Debezium PostgreSQL connector to monitor the changes committed
on a PostgreSQL server, first install the [logical decoding plugin](https://debezium.io/docs/install/postgres-plugins/) into the PostgreSQL
server. Enable a replication slot and configure a user with sufficient
privileges to perform the replication.

To monitor a PostgreSQL database running in [Amazon RDS](https://aws.amazon.com/rds/), refer to the Debezium documentation for
[PostgreSQL on AmazonRDS](https://debezium.io/docs/connectors/postgresql/#amazon-rds).

<a id="enable-logical-decoding"></a>

### Enable logical decoding and replication on the PostgreSQL server

The Postgres relational database management system has a feature called logical
decoding that allows clients to extract all persistent changes to database
tables into a coherent format. This formatted data can be interpreted without
detailed knowledge of the internal state of the database. An output plugin
transforms the data from the write-ahead log’s internal representation into a
format the consumer of a replication slot needs.

The Debezium PostgreSQL connector works with one of the following supported
logical decoding plugins from Debezium:

- [protobuf](https://github.com/debezium/postgres-decoderbufs/blob/master/README.md) : To encode changes in Protobuf format
- [wal2json](https://github.com/eulerto/wal2json/blob/master/README.md) : To encode changes in JSON format

#### Install the `wal2json` plugin

Before executing the commands, make sure the user has write-privilege to the
`wal2json` library at the PostgreSQL lib directory. Note that for the test
environment, this directory is `/usr/pgsql-9.6/lib/`. In the test environment
set the export path as shown below:

```none
export PATH="$PATH:/usr/pgsql-9.6/bin"
```

Enter the **wal2json** installation commands.

```none
git clone https://github.com/eulerto/wal2json -b master --single-branch \
&& cd wal2json \
&& git checkout 92b33c7d7c2fccbeb9f79455dafbc92e87e00ddd \
&& make && make install \
&& cd .. \
&& rm -rf wal2json
```

#### Enable replication on the PostgreSQL server

Add the following lines to the end of the
`/usr/share/postgresql/postgresql.conf` PostgreSQL configuration file. These
lines include the plugin at the shared libraries and adjust some Write-Ahead Log
(WAL) and streaming replication settings.

```none
# LOGGING
log_min_error_statement = fatal
# CONNECTION
listen_addresses = '*'
# MODULES
shared_preload_libraries = 'decoderbufs'
# REPLICATION
wal_level = logical             # minimal, archive, hot_standby, or logical (change requires restart)
max_wal_senders = 1             # max number of walsender processes (change requires restart)
#wal_keep_segments = 4          # in logfile segments, 16MB each; 0 disables
#wal_sender_timeout = 60s       # in milliseconds; 0 disables
max_replication_slots = 1       # max number of replication slots (change requires restart)
```

Note that in the PostgreSQL configuration file you may also set
`shared_preload_libraries` to the following value:

```text
...
# MODULES
shared_preload_libraries = 'wal2json'
...
```

#### Initialize replication permissions

Add the following lines to the end of the `pg_hba.conf` PostgreSQL configuration file. These lines configure the client authentication for the database replication.

```none
############ REPLICATION ##############
local   replication     postgres                          trust
host    replication     postgres  127.0.0.1/32            trust
host    replication     postgres  ::1/128                 trust
```

<a id="postgres-source-connector-quickstart"></a>

## Quick start

The [Debezium PostgreSQL Connector](https://debezium.io/docs/connectors/postgresql/)  is a source connector that
can record events for each table in a separate Kafka topic, where they can be
easily consumed by applications and services.

For an example of how to get Kafka Connect connected to [Confluent Cloud](/cloud/current/index.html), see
[Connect Self-Managed Kafka Connect to Confluent Cloud](/cloud/current/cp-component/connect-cloud-config.html#distributed-cluster).

### Install the connector

Refer to the [Debezium tutorial](https://github.com/debezium/debezium-examples/tree/master/tutorial#using-postgres)
if you want to use Docker images to set up Kafka, ZooKeeper and Connect. For the
following tutorial, you need to have a local Confluent Platform installation. Note that as of
Confluent Platform 7.5, ZooKeeper is deprecated for new deployments. Confluent recommends KRaft
mode for new deployments.

1. Navigate to your Confluent Platform installation directory and run the following command to
   install the connector:
   ```bash
   confluent connect plugin install debezium/debezium-connector-postgresql:<version-number>
   ```

   Note that adding a new connector plugin requires restarting Kafka Connect.
   Use the Confluent CLI to restart Kafka Connect by running the two commands:
   ```bash
   confluent local services connect stop
   confluent local services connect start
   ```

   You should see output similar to the following:
   ```text
   Using CONFLUENT_CURRENT: /Users/username/Sandbox/confluent-snapshots/var/confluent.NuZHxXfq
   Starting Zookeeper
   Zookeeper is [UP]
   Starting Kafka
   Kafka is [UP]
   Starting Schema Registry
   Schema Registry is [UP]
   Starting Kafka REST
   Kafka REST is [UP]
   Starting Connect
   Connect is [UP]
   ```
2. Check if the PostgreSQL plugin has been installed correctly and picked
   up by the plugin loader.
   ```bash
   curl -sS localhost:8083/connector-plugins | jq '.[].class' | grep postgres
   "io.debezium.connector.postgresql.PostgresConnector"
   ```

### Set up PostgreSQL using Docker (optional)

If you do not have a native installation of PostgreSQL, you may use the
following command to start a new container to run a PostgreSQL database server
preconfigured with the [logical decoding](https://www.postgresql.org/docs/9.6/logicaldecoding-explanation.html)
plugin, [replication slot](https://www.postgresql.org/docs/9.6/logicaldecoding-walsender.html) and an
`inventory` test database.

```bash
# Pull docker image
docker pull debezium/example-postgres

# Run docker container
docker run -it --rm --name postgres -p 5432:5432 \
-e POSTGRES_USER=postgres -e POSTGRES_PASSWORD=postgres \
debezium/example-postgres

# In a separate terminal, launch psql to run SQL queries:
docker run -it --rm --name psql_client \
-e PGOPTIONS="--search_path=inventory" \
-e PGPASSWORD=postgres --link postgres:postgres debezium/example-postgres \
psql -h postgres -U postgres

# To see the list of relations in the inventory database, type \d at the postgres prompt. To exit, type \q
```

### Enable logical decoding on the PostgreSQL server

Logical encoding is already enabled if you set up PostgreSQL using the Docker image (in the previous section).
On your native installation, follow these steps to  [Enable logical decoding and replication on the PostgreSQL server](#enable-logical-decoding).

### Start the Debezium PostgreSQL connector

1. Create the file `register-postgres.json` to store the following connector
   configuration:
   ```json
   {
      "name": "inventory-connector",
      "config": {
         "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
         "tasks.max": "1",
         "database.hostname": "0.0.0.0",
         "database.port": "5432",
         "database.user": "postgres",
         "database.password": "postgres",
         "database.dbname" : "postgres",
         "topic.prefix": "dbserver1",
         "schema.include.list": "inventory"
         }
   }
   ```
2. Start the connector.
   ```bash
   curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors/ -d @register-postgres.json
   ```

### Start your Kafka consumer

Start the consumer in a new terminal session.

```bash
confluent local consume dbserver1.inventory.customers --from-beginning
```

When you enter SQL queries in bash (to add or modify records in the database) messages populate and are displayed on your consumer terminal to reflect those records.

The following is an example psql query to update a record in the customers table.

```bash
update customers set first_name = 'Sarah' where id = 1001;
```

### Clean up resources

1. Delete the connector and stop Confluent services.
   ```bash
   curl -X DELETE localhost:8083/connectors/inventory-connector
   confluent local stop
   ```
2. Stop PostgreSQL containers.
   ```bash
   docker stop psql_client # Alternatively type \q at the psql prompt
   docker stop postgres
   ```

#### NOTE
Portions of the information provided here derive from documentation
originally produced by the [Debezium Community](https://debezium.io/).
Work produced by Debezium is licensed under [Creative Commons 3.0](https://creativecommons.org/licenses/by/3.0/).