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
   • User: 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. To enable pg-user to create a publication, assign them the mdb_replication role.

2. Connect to the db1 database under 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. Topic names are generated based on the same conventions that are used 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 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 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 <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 a list of hosts in the Managed Service for Apache Kafka® cluster.

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

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

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.