Discuss with expertTry it for free
© 2026 Direct Cursus Technology L.L.C.

Delivering data to Yandex Managed Service for Apache Kafka® using Yandex Data Transfer

Written by
Updated at July 22, 2026
View in Markdown

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

To set up CDC using Data Transfer:

  1. Set up your infrastructure.
  2. Prepare the source cluster.
  3. Set up the target cluster.
  4. Prepare and activate your transfer.
  5. Test the transfer.

If you no longer need the resources you created, delete them.

Getting started

Sign up for Yandex Cloud and create a billing account:

  1. Navigate to the management console and log in to Yandex Cloud or create a new account.
  2. On the Yandex Cloud Billing page, make sure you have a billing account linked and it has the ACTIVE or TRIAL_ACTIVE status. If you do not have a billing account, create one and link a cloud to it.

If you have an active billing account, you can create or select a folder for your infrastructure on the cloud page.

Learn more about clouds and folders here.

Set up your infrastructure

Note

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.

  1. Create a Managed Service for PostgreSQL source cluster with any suitable configuration, using the following settings:

    • Database: db1.

    • User: pg-user.

    • Hosts: Publicly available.

  2. Create a Managed Service for Apache Kafka® target cluster in any suitable configuration with publicly accessible hosts.

  3. If using security groups, configure them to allow internet access to your clusters:

  1. If you do not have Terraform yet, install it.

  2. Get the authentication credentials. You can add them to environment variables or specify them later in the provider configuration file.

  3. Configure and initialize a provider. There is no need to create a provider configuration file manually, you can download it.

  4. Place the configuration file in a separate working directory and specify the parameter values. If you did not add the authentication credentials to environment variables, specify them in the configuration file.

  5. Download the cdc-mpg-mkf.tf configuration file to the same working directory.

    This file describes:

    • Network.
    • Subnet in the ru-central1-a availability zone.
    • Security group for the Managed Service for PostgreSQL cluster and rules for access to the cluster from the internet.
    • Security group for the Managed Service for Apache Kafka® cluster and rules for access to the cluster from the internet.
    • Managed Service for PostgreSQL cluster with public internet access.
    • Database and user with the mdb_replication role in the Managed Service for PostgreSQL cluster.
    • Managed Service for Apache Kafka® cluster with public internet access.
    • Topic and user with the ACCESS_ROLE_CONSUMER and ACCESS_ROLE_PRODUCER roles for this topic in the Managed Service for Apache Kafka® cluster.
    • Source and target endpoints.
    • Transfer.

  6. Specify the following in the configuration file:

    • network_name: Network name.
    • subnet_name: Subnet name.
    • pg_sg_name: Name of security group for the Managed Service for PostgreSQL cluster.
    • kf_sg_name: Name of security group for the Managed Service for Apache Kafka® cluster.
    • pg_cluster_version: Managed Service for PostgreSQL cluster version.
    • pg_cluster_name: Managed Service for PostgreSQL cluster name.
    • pg_password: User password in the Managed Service for PostgreSQL cluster.
    • kf_cluster_version: Managed Service for Apache Kafka® cluster version.
    • kf_cluster_name: Managed Service for Apache Kafka® cluster name.
    • kf_password: User password in the Managed Service for Apache Kafka® cluster.
    • source_endpoint_name: Source endpoint name.
    • target_endpoint_name: Target endpoint name.
    • transfer_name: Transfer name.
    • transfer_enabled = 0: Disables the creation of endpoints and transfers. They will be created during the preparation of the transfer.

  7. Make sure the Terraform configuration files are correct using this command:

    terraform validate

    Terraform will display any configuration errors detected in your files.

  8. Create the required infrastructure:

    1. Run this command to view the planned changes:

      terraform plan

      If you described the configuration correctly, the terminal will display a list of the resources to update and their parameters. This is a verification step that does not apply changes to your resources.

    2. If everything looks correct, apply the changes:

      1. Run this command:

        terraform apply

      2. Confirm updating the resources.

      3. Wait for the operation to complete.

    All the required resources will be created in the specified folder. You can check resource availability and their settings in the management console.

Prepare the source cluster

  1. If you created the infrastructure manually, assign the mdb_replication role to pg-user. The user needs this role to be able to create a publication through which Data Transfer will receive information about changes in the source cluster data.

  2. Connect to the db1 database as pg-user.

  3. 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);

Prepare the target cluster

If you created the infrastructure using Terraform, skip this step and proceed to transfer preparation and activation.

The target cluster's setup will vary depending on the topic management method you are using. Topic names follow the same convention as Debezium: <topic_prefix>.<schema_name>.<table_name>. In this tutorial, we will use the cdc example prefix.

If you manage topics using the Yandex Cloud interfaces (management console, CLI, API):

  1. Create a topic named cdc.public.measurements.

    If you need to track data changes in multiple tables, create a separate topic for each one.

  2. Create a user named kafka-user with the ACCESS_ROLE_CONSUMER and ACCESS_ROLE_PRODUCER roles for the new topic.

If you manage topics using the Kafka Admin API:

  1. Create an admin user named kafka-user.

  2. In addition to the ACCESS_ROLE_ADMIN role, assign the admin user the ACCESS_ROLE_CONSUMER and ACCESS_ROLE_PRODUCER roles for topics whose names begin with the cdc prefix.

    The system will automatically create the required topics when the first change occurs in the monitored tables of the source cluster. While this approach can be convenient for tracking changes across multiple tables, it requires reserving free storage space in your cluster. For more information, see Storage in Managed Service for Apache Kafka®.

Prepare and activate a transfer

  1. Create endpoints.

    • Source endpoint:

      • Database type: PostgreSQL.
      • Endpoint parameters:
        • Connection settings: Managed Service for PostgreSQL cluster.
        • Managed Service for PostgreSQL cluster: Your previously created Managed Service for PostgreSQL cluster.
        • Database: db1.
        • User: pg-user.
        • Password: pg-user password.
        • Included tables: public.measurements.

    • Target endpoint:

      • Database type: Kafka.

      • Endpoint parameters:

        • Connection type: Managed Service for Apache Kafka cluster.

          • Managed Service for Apache Kafka cluster: Select the target cluster.
          • Authentication: Specify the kafka-user credentials.

        • Topic: Topic full name.

        • Topic full name: cdc.public.measurements.

        If you need to track changes in multiple tables, fill in the fields as follows:

        • Topic: Topic prefix.
        • Topic prefix: Enter the cdc prefix you used to generate topic names.

  2. Create a transfer with the following settings:

    • Endpoints:
      • Source: Source endpoint you created earlier.
      • Target: Target endpoint you created earlier.
    • Transfer type: Replication.

  3. Activate the transfer and wait for its status to change to Replicating.

  1. Specify transfer_enabled = 1 in the cdc-mpg-mkf.tf file.

  2. Make sure the Terraform configuration files are correct using this command:

    terraform validate

    Terraform will display any configuration errors detected in your files.

  3. Create the required infrastructure:

    1. Run this command to view the planned changes:

      terraform plan

      If you described the configuration correctly, the terminal will display a list of the resources to update and their parameters. This is a verification step that does not apply changes to your resources.

    2. If everything looks correct, apply the changes:

      1. Run this command:

        terraform apply

      2. Confirm updating the resources.

      3. Wait for the operation to complete.

    Endpoints and a transfer will be created. The transfer will be activated automatically as soon as it is created.

  4. Wait for the transfer status to change to Replicating.

    You can check the transfer status in the management console.

Test the transfer

  1. Install the kcat (kafkacat) utility and the PostgreSQL command-line client on your local machine. For example, on Ubuntu 20.04, use this command:

    sudo apt update && sudo apt install kafkacat postgresql-client --yes

    Check that you can use it to connect to the Managed Service for Apache Kafka® source cluster over SSL.

  2. In a separate terminal, run kafkacat in consumer mode:

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

    You can get the FQDNs of broker hosts with the list of hosts in the Managed Service for Apache Kafka® cluster.

    The utility will print the public.measurements table schema along with the data from previously inserted rows.

    Message snippet example
    {
  "payload": {
    "consumer":"dttuhfpp97l3********"
  },
  "schema": {
    "fields": [
      {
        "field": "consumer",
        "optional":false,
        "type":"string"
      }
    ],
    "name": "__data_transfer_stub.public.__consumer_keeper.Key",
    "optional":false,
    "type":"struct"
  }
}:{
  "payload": {
    "after": {
      "consumer":"dttuhfpp97l3********l",
      "locked_by":"dttuhfpp97l3********-1",
      "locked_till":"2022-05-15T09:55:18Z"
    },
  "before": null,
  "op":"u",
  "source": {
    "connector":"postgresql",
    "db":"db1",
    "lsn":85865797008,
    "name":"__data_transfer_stub",
    "schema":"public",
    "snapshot":"false",
    "table":"__consumer_keeper",
    "ts_ms":1652608518883,
    "txId":245165,
    "version":"1.1.2.Final",
    "xmin":null
  },
...

  3. Connect to the source cluster and populate the measurements table with data:

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

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

Delete the resources you created

Some resources are not free of charge. Delete the resources you no longer need to avoid paying for them:

  1. Deactivate and delete the transfer.

  2. Delete the endpoints.

  3. Delete the clusters:

  4. If static public IP addresses were used for accessing the cluster hosts, release and delete them.

  1. In the terminal window, go to the directory containing the infrastructure plan.

    Warning

    Make sure the directory has no Terraform manifests with the resources you want to keep. Terraform deletes all resources that were created using the manifests in the current directory.

  2. Delete resources:

    1. Run this command:

      terraform destroy

    2. Confirm deleting the resources and wait for the operation to complete.

    All the resources described in the Terraform manifests will be deleted.

Previous
Setting up a connection from a container in Serverless Containers
Next
Delivering data to Yandex Managed Service for YDB using Yandex Data Transfer