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

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

Required paid resourcesRequired paid resources

• Managed Service for PostgreSQL cluster: Computing resources allocated to hosts, storage and backup size (see Managed Service for PostgreSQL pricing).
• Managed Service for Apache Kafka® cluster: Computing resources allocated to hosts, storage and backup size (see Managed Service for Apache Kafka® pricing).
• Public IP addresses if public access is enabled for cluster hosts (see Virtual Private Cloud pricing).
• Each transfer: Use of computing resources and number of transferred data rows (see Data Transfer pricing).

Getting startedGetting started

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 using any suitable configuration with publicly accessible hosts.

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

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

4. 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 can use it to connect to the Managed Service for Apache Kafka® source cluster over SSL.

Set up the source clusterSet up the source cluster

1. For Data Transfer to receive data change notifications from a Managed Service for PostgreSQL cluster, you must create a publication on the source cluster. Assign pg-user the mdb_replication role to allow publication creation.

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

Set up the target clusterSet up the target cluster

The configuration may vary depending on the chosen topic management method. Topic names follow the same convention as Debezium: <topic_prefix>.<schema_name>.<table_name>. In this tutorial, we will use the cdc example prefix.

Set up and activate the transferSet up and activate the 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: Specify 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.

Test your transferTest your transfer

1. 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
     },
   ...
   

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. Check that the terminal running kafkacat shows the information about the new row.

Delete the resources you createdDelete the resources you created

To reduce the consumption of resources you do not need, delete them:

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 you used static public IP addresses to access the cluster hosts, release and delete them.