Delivering data from Yandex Managed Service for PostgreSQL to Yandex Managed Service for Apache Kafka® using Debezium

Written by

Updated at February 18, 2026

Required paid resources
Getting started
Set up the source cluster
Configure Debezium
Set up the target cluster
Run the Debezium connector
Check that Debezium works properly
Delete the resources you created

You can track data changes in Managed Service for PostgreSQL and send them to Managed Service for Apache Kafka® using change data capture (CDC).

In this tutorial, you will learn how to create a virtual machine in Yandex Cloud and set up Debezium, an open-source software framework for CDC.

Required paid resources

The support cost for this solution includes:

Managed Service for PostgreSQL cluster fee: Covers the use of computing resources allocated to hosts and disk storage (see Managed Service for PostgreSQL pricing).
Managed Service for Apache Kafka® cluster fee: Covers the use of computing resources allocated to hosts (including ZooKeeper hosts) and disk storage (see Apache Kafka® pricing).
VM fee: Covers the use of computing resources, the OS, and the storage (see Compute Cloud pricing).
Fee for using public IP addresses for the VM and hosts of the two clusters (see Virtual Private Cloud pricing).

Public access to cluster hosts is required if you plan to connect to the cluster via the internet. This connection option is simpler and is recommended for the purposes of this guide. You can connect to non-public hosts as well but only from Yandex Cloud virtual machines located in the same cloud network as the cluster.

Create a source cluster with the following settings:
- Hosts: Publicly available
- Database: db1
- User: user1
Create a Managed Service for Apache Kafka® target cluster in any suitable configuration with publicly accessible hosts.
Create a virtual machine running Ubuntu 20.04 with a public IP address.
If you use security groups, configure them to allow connections to the clusters from the internet and from the VM you created, and to allow SSH access to that VM from the internet:
- Configuring Managed Service for Apache Kafka® cluster security groups.
- Configuring Managed Service for PostgreSQL cluster security groups.
Connect to your VM over SSH and complete its initial setup:
1. Install the dependencies:
```
sudo apt update && \
    sudo apt install kafkacat openjdk-17-jre postgresql-client --yes
```
  Make sure you can use it to connect to the Managed Service for Apache Kafka® source cluster over SSL.
2. Create a directory for Apache Kafka®:
```
sudo mkdir -p /opt/kafka/
```
3. Download the archive with Apache Kafka® executables and unpack it into this directory. For example, to download and unpack Apache Kafka® 3.0, run this command:
```
wget https://archive.apache.org/dist/kafka/3.0.0/kafka_2.13-3.0.0.tgz && \
sudo tar xf kafka_2.13-3.0.0.tgz --strip 1 --directory /opt/kafka/
```
  You can check the current Apache Kafka® version on the project’s download page.
4. Install certificates on the VM and make sure you can access the clusters:
  - Managed Service for Apache Kafka® (use kafkacat).
  - Managed Service for PostgreSQL (use psql).
5. Create a directory to store the files required for the Debezium connector:
```
sudo mkdir -p /etc/debezium/plugins/
```
6. To enable the Debezium connector to connect to Managed Service for Apache Kafka® broker hosts, add the SSL certificate to Java Key Store. For extra storage security, specify a password of at least six characters in the -storepass setting:
```
sudo keytool \
    -importcert \
    -alias YandexCA -file /usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt \
    -keystore /etc/debezium/keystore.jks \
    -storepass <JKS_password> \
    --noprompt
```

Set up the source cluster

Assign the mdb_replication role to user1.

This is necessary to create a publication for Debezium to monitor changes in a Managed Service for PostgreSQL cluster.
Connect to the db1 database as user1.

Populate the database with test data. In this example, we will use a simple table with car sensor information.

Create a table:

CREATE TABLE public.measurements (
    "device_id" text PRIMARY KEY NOT NULL,
    "datetime" timestamp NOT NULL,
    "latitude" real NOT NULL,
    "longitude" real NOT NULL,
    "altitude" real NOT NULL,
    "speed" real NOT NULL,
    "battery_voltage" real,
    "cabin_temperature" real NOT NULL,
    "fuel_level" real
);

Populate the table with data:

INSERT INTO public.measurements VALUES
    ('iv9a94th6rzt********', '2020-06-05 17:27:00', 55.70329032, 37.65472196,  427.5,    0, 23.5, 17, NULL),
    ('rhibbh3y08qm********', '2020-06-06 09:49:54', 55.71294467, 37.66542005, 429.13, 55.5, NULL, 18, 32),
    ('iv9a94th678t********', '2020-06-07 15:00:10', 55.70985913, 37.62141918,  417.0, 15.7, 10.3, 17, NULL);

Create a publication for the table:

CREATE PUBLICATION mpg_publication FOR TABLE public.measurements;

Configure Debezium

Connect to the VM over SSH.

Download the current Debezium connector and unpack into the /etc/debezium/plugins/ directory.

You can check the current connector version on the project page. Below are commands for 1.9.4.Final.

VERSION="1.9.4.Final"
wget https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/${VERSION}/debezium-connector-postgres-${VERSION}-plugin.tar.gz && \
sudo tar -xzvf debezium-connector-postgres-${VERSION}-plugin.tar.gz -C /etc/debezium/plugins/

Create a file named /etc/debezium/mdb-connector.conf with the Debezium connector settings for connecting to the source cluster:
```
name=debezium-mpg
connector.class=io.debezium.connector.postgresql.PostgresConnector
plugin.name=pgoutput
database.hostname=c-<cluster_ID>.rw.mdb.yandexcloud.net
database.port=6432
database.user=user1
database.password=<user1_password>
database.dbname=db1
database.server.name=mpg
table.include.list=public.measurements
publication.name=mpg_publication
slot.name=debezium_slot
heartbeat.interval.ms=15000
heartbeat.topics.prefix=debezium-heartbeat
snapshot.mode=always
```
Where:
- name: Logical name of the Debezium connector. It is used for the connector’s internal needs.
- database.hostname: Special FQDN for connection to the source cluster's master host.
  
  You can get the cluster ID from the list of clusters in your folder.
- database.user: PostgreSQL username.
- database.dbname: PostgreSQL database name.
- database.server.name: Name of the database server that Debezium will use when choosing a topic for sending messages.
- table.include.list: Names of tables for Debezium to track changes in. Specify full names that include the schema name (default: public). Debezium will use values from this field when choosing a topic for sending messages.
- publication.name: Name of the publication created on the source cluster.
- slot.name: Name of the replication slot Debezium will create for the publication.
- heartbeat.interval.ms and heartbeat.topics.prefix: Heartbeat settings required for Debezium.
- snapshot.mode: Type of snapshot created at connector startup. For the connector to run properly, set this parameter to always.

Set up the target cluster

Create a topic to store data from the source cluster:
- Name: mpg.public.measurements.
  
  Data topic names follow the <server_name>.<schema_name>.<table_name> convention.
  
  According to the Debezium configuration file:
  - The mpg server name is specified in the database.server.name parameter.
  - The public schema name is specified together with the measurements table name in the table.include.list parameter.
If you need to track data changes in multiple tables, create a separate topic for each one.
Create a service topic for tracking the connector status:
- Name: debezium-heartbeat.mpg
  
  Service topic names follow the <prefix_for_heartbeat>.<server_name> convention.
  
  According to the Debezium configuration file:
  - The debezium-heartbeat prefix is specified in the heartbeat.topics.prefix parameter.
  - database.server.name specifies the server name, mpg.
- Cleanup policy: Compact.
If you need to capture data from multiple source clusters, create a separate service topic for each one.
Create a user named debezium.
Grant debezium the ACCESS_ROLE_CONSUMER and ACCESS_ROLE_PRODUCER permissions for the topics you created.

Run the Debezium connector

Create a file with Debezium worker settings:

/etc/debezium/worker.conf

# AdminAPI connect properties
bootstrap.servers=<broker_host_1_FQDN>:9091,...,<broker_host_N_FQDN>:9091
sasl.mechanism=SCRAM-SHA-512
security.protocol=SASL_SSL
ssl.truststore.location=/etc/debezium/keystore.jks
ssl.truststore.password=<JKS_password>
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="debezium" password="<debezium_user_password>";

# Producer connect properties
producer.sasl.mechanism=SCRAM-SHA-512
producer.security.protocol=SASL_SSL
producer.ssl.truststore.location=/etc/debezium/keystore.jks
producer.ssl.truststore.password=<JKS_password>
producer.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="debezium" password="<debezium_user_password>";

# Worker properties
plugin.path=/etc/debezium/plugins/
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true
offset.storage.file.filename=/etc/debezium/worker.offset

In a separate terminal, run the connector:

sudo /opt/kafka/bin/connect-standalone.sh \
    /etc/debezium/worker.conf \
    /etc/debezium/mdb-connector.conf

Check that Debezium works properly

In a separate terminal, run kafkacat in consumer mode:

kafkacat \
    -C \
    -b <broker_host_1_FQDN>:9091,...,<broker_host_N_FQDN>:9091 \
    -t mpg.public.measurements \
    -X security.protocol=SASL_SSL \
    -X sasl.mechanisms=SCRAM-SHA-512 \
    -X sasl.username=debezium \
    -X sasl.password=<password> \
    -X ssl.ca.location=/usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt \
    -Z \
    -K:

The output will return the data format schema of the db1.public.measurements table and information about the previously added rows.

Message snippet example

{
"schema": {
    ...
},
"payload": {
    "before": null,
    "after": {
        "device_id": "iv9a94th6rzt********",
        "datetime": 1591378020000000,
        "latitude": 55.70329,
        "longitude": 37.65472,
        "altitude": 427.5,
        "speed": 0.0,
        "battery_voltage": 23.5,
        "cabin_temperature": 17.0,
        "fuel_level": null
    },
    "source": {
        "version": "1.8.1.Final",
        "connector": "postgresql",
        "name": "mpg",
        "ts_ms": 1628245046882,
        "snapshot": "true",
        "db": "db1",
        "sequence": "[null,\"4328525512\"]",
        "schema": "public",
        "table": "measurements",
        "txId": 8861,
        "lsn": 4328525328,
        "xmin": null
    },
    "op": "r",
    "ts_ms": 1628245046893,
    "transaction": null
  }
}

Connect to the source cluster.

When connecting, you may get this error: ERROR Postgres roles LOGIN and REPLICATION are not assigned to user. You can ignore it, as it does not affect Debezium performance.

Add another row to the measurements table:

INSERT INTO public.measurements VALUES ('iv7b74th678t********', '2020-06-08 17:45:00', 53.70987913, 36.62549834, 378.0, 20.5, 5.3, 20, NULL);

Check that the terminal running kafkacat shows the information about the new row.

Delete the resources you created

Delete the resources you no longer need to avoid paying for them:

Delete the virtual machine.

If you reserved a public static IP address for the virtual machine, release and delete it.
Delete the clusters:
- Managed Service for Apache Kafka®.
- Managed Service for PostgreSQL.

Delivering data from Yandex Managed Service for PostgreSQL to Yandex Managed Service for Apache Kafka® using Debezium

Required paid resourcesRequired paid resources

Getting startedGetting started

Set up the source clusterSet up the source cluster

Configure DebeziumConfigure Debezium

Set up the target clusterSet up the target cluster

Run the Debezium connectorRun the Debezium connector

Check that Debezium works properlyCheck that Debezium works properly

Delete the resources you createdDelete the resources you created

Was the article helpful?