PostgreSQL change data capture and delivery to Apache Kafka®

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. Prepare the source cluster.
2. Prepare the target cluster.
3. Prepare and activate the transfer.
4. Test the transfer.

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

Getting startedGetting started

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

   • Database: db1
   • Username: pg-user
   • Hosts: Publicly available

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

3. If you are using security groups, configure them to enable connecting to the clusters from the internet:

   • Guide for Managed Service for PostgreSQL
   • Guide for Managed Service for Apache Kafka®

4. Install the kcat (kafkacat) utility and PostgreSQL command-line client on the local machine. For example, in Ubuntu 20.04, run:

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

Prepare the source clusterPrepare the source cluster

1. For Data Transfer to get notifications of data changes from a Managed Service for PostgreSQL cluster, create a publication in a source cluster. The pg-user user can create a publication after you assign them the mdb_replication role.

2. Connect to the db1 database on behalf of pg-user.

3. Add test data to the database. In this example, a simple table with information from car sensors is used.

   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 clusterPrepare the target cluster

The settings vary depending on the topic management method used. The same conventions are used for topic names as in Debezium: <topic_prefix>.<schema_name>.<table_name>. In this tutorial, the cdc prefix is used as an example.

Prepare and activate the transferPrepare and activate the transfer

1. Create an endpoint.

   • Source endpoint:

     • Database type: PostgreSQL.
     • Endpoint parameters:
       • Connection settings: Managed Service for PostgreSQL cluster.
       • Managed Service for PostgreSQL cluster: Select the created Managed Service for PostgreSQL cluster.
       • Database: db1.
       • User: pg-user.
       • Password: Enter the password for pg-user.
       • 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: Enter the details of the created kafka-user user.

       • Topic: Topic full name.

       • Topic full name: cdc.public.measurements.

       If you need to track changes in multiple tables, fill out 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: Created source endpoint.
     • Target: Created target endpoint.
   • Transfer type: Replication.

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

Test the transferTest the transfer

1. In a separate terminal, run the kafkacat utility in consumer mode:

   kafkacat \
       -C \
       -b <FQDN_of_broker_host_1>:9091,...,<FQDN_of_broker_host_N>: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 obtain the FQDNs of broker hosts with a list of hosts in the Managed Service for Apache Kafka® cluster.

   The data format schema of the public.measurements table and the information about the previously added rows will be printed.

   Example of the message fragment

   {
     "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
     },
   ...
   

2. Connect to the source cluster and add data 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);
   

3. Make sure the terminal running kafkacat displays details about the added row.

Delete the resources you createdDelete the resources you created

Some resources are not free of charge. To avoid paying for them, delete the resources you no longer need:

1. Deactivate and delete the transfer.

2. Delete the endpoints.

3. Delete the clusters:

   • Managed Service for Apache Kafka®.
   • Managed Service for PostgreSQL.

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