Creating Replication Slots to Enable CDC

The following uses the testdb database as an example.

1. Log in to the management console.
2. Click in the upper left corner and select a region.
3. Click in the upper left corner of the page and choose Databases > Relational Database Service.
4. On the Instances page, click the instance name.
5. On the Databases page, click Create Database. In the displayed dialog box, configure required parameters and click OK.
   Figure 1 Creating a database
   

The account created in this example is for reference only. You can change it as required.

1. On the Accounts page, click Create Account.
2. In the displayed dialog box, enter the username, password, permissions, and remarks, and click OK.
3. Create a standard account, for example, cdc_user.
4. Connect to the RDS for PostgreSQL instance using the account root.

   psql -h <instance-address> -p 5432 -U root -d testdb

5. Assign the Replication role to the account cdc_user and query the result:

   ALTER USER cdc_user WITH REPLICATION;
   SELECT rolreplication FROM  pg_roles WHERE rolname='cdc_user';

   Query result:

   rolreplication
   ----------------
    t
   (1 row)

6. Grant permissions to the cdc_user account:

   GRANT SELECT ON ALL TABLES IN SCHEMA PUBLIC to cdc_user;

1. Query the instance's parameter settings:

   SELECT name,
          setting,
          short_desc,
          source
   FROM pg_settings
   WHERE name ='wal_level';

   Query result:

   name         		   | setting |                               short_desc                                |       source
   -----------------------+---------+-------------------------------------------------------------------------+--------------------
    wal_level             | replica | Sets the level of information written to the WAL.                       | configuration file
   (1 rows)

   wal_level specifies how much information is written to the WAL. It determines the WAL's purpose (such as crash recovery, physical replication, or logical replication). The following are three main values of wal_level, from least to most detailed:

   • minimal: The system records only the information required to recover from a crash. No replication is supported. (This value is almost not used in production environments.)
   • replica (default): Enough data needed for physical replication is written. This level supports physical replication (such as, primary-standby synchronization, where the standby instance replays WAL logs to remain synchronous with the primary instance) and allows a read replica to function as a hot standby.
   • logical: The system records the data logged at replica level and also adds the metadata (such as row-level change data and table structure information) necessary for logical replication. This setting supports logical replication and CDC.
2. On the Instances page, click the instance name.
3. In the navigation pane, choose Parameters.
4. Change the value of wal_level to logical.

   For details about how to modify instance parameters, see Modifying Parameters of an RDS for PostgreSQL Instance.

   To prevent impact on workloads, modify parameters during off-peak hours.

5. After changing the parameter value, reboot your RDS for PostgreSQL instance to apply it.

Instance parameters have been modified in Step 3: Modify RDS for PostgreSQL Instance Parameters, so perform the following operations after the RDS for PostgreSQL instance is rebooted and its status changes to Available.

1. Connect to the RDS for PostgreSQL instance using the account root.

   psql -h <instance-address> -p 5432 -U root -d testdb

2. Create a replication slot named cdc_replication_slot with the decoding plugin test_decoding:

   SELECT * FROM pg_create_logical_replication_slot(
       'cdc_replication_slot',
       'test_decoding'
   );

   After the execution, the replication slot name and initial LSN are displayed in the command output, as shown in the following:

   	slot_name    |    lsn
    ----------------+-----------
    cdc_replication_slot | 0/16A00000
   (1 row)

3. List all replication slots to verify that the replication slot is created:

   SELECT slot_name, plugin, active FROM pg_replication_slots;

   If slot_name is cdc_replication_slot and plugin is test_decoding, the replication slot is created.

   slot_name             |    plugin     | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn | wal_status | safe_wal_size  | two_phase
   ----------------------+---------------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------+------------+----------------+-----------
    cdc_replication_slot | test_decoding | logical   |  31721 | testdb   | f         | f      |            |      |        25523| 2/2500A1B0  | 2/25003882          | reserved   |                | f
   (1 row)

Create test data to simulate a production environment.

1. Create a test table for generating change data.

   CREATE TABLE public.tb_tests (
       id SERIAL PRIMARY KEY,
       product_name VARCHAR(50) NOT NULL,
       price NUMERIC(10,2) NOT NULL,
       create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
   );

2. Insert data.

   INSERT INTO public.tb_tests (product_name, price) VALUES ('Laptop', 5999.99);
   INSERT INTO public.tb_tests (product_name, price) VALUES ('Mechanical keyboard', 299.99);

3. Update data.

   UPDATE public.tb_tests SET price = 5899.99 WHERE id = 1;

4. Delete data.

   DELETE FROM public.tb_tests  WHERE id = 2;

1. Connect to the RDS for PostgreSQL instance as the user cdc_user.

   psql -h <instance-address> -p 5432 -U cdc_user -d testdb

2. Read data from the replication slot:

   SELECT * FROM pg_logical_slot_peek_changes(
       'cdc_replication_slot', --Replication slot name
       NULL,               --Start LSN (NULL indicates that the data is read from the current position.)
       NULL,               --Maximum number of returned records (NULL indicates that the number is not limited.)
      );

   Example output:

   lsn         |  xid  |                                  data
   ------------+-------+-------------------------------------------------------------------------
    0/16A01200 | 1234 | BEGIN 1234
    0/16A01200 | 1234 | table public.tb_tests: INSERT: id[integer]:1 product_name[character varying]:'Laptop' price[numeric(10,2)]:5999.99 create_time[timestamp with time zone]:'2024-05-20 10:00:00+08'
    0/16A01350 | 1234 | table public.tb_tests: INSERT: id[integer]:2 product_name[character varying]:'Mechanical keyboard' price[numeric(10,2)]:299.99 create_time[timestamp with time zone]:'2024-05-20 10:01:00+08'
    0/16A014A0 | 1234 | COMMIT 1234
    0/16A015F0 | 1235 | BEGIN 1235
    0/16A015F0 | 1235 | table public.tb_tests: UPDATE: id[integer]:1 price[numeric(10,2)]:5899.99 (old:5999.99)
    0/16A016E0 | 1235 | COMMIT 1235
    0/16A017D0 | 1236 | BEGIN 1236
    0/16A017D0 | 1236 | table public.tb_tests: DELETE: id[integer]:2
    0/16A01860 | 1236 | COMMIT 1236
    (8 rows)

On the client, run the following command to connect to the DB instance and consume data from the replication slot cdc_replication_slot:

pg_recvlogical \
-h <PostgreSQL-instance-address> \
  -p 5432 \                  
  -U cdc_user \              
  -d testdb \                
  --slot=cdc_replication_slot \   
  --start \                  
  -f -                       

Specify the parameter values as required:

• -h: RDS for PostgreSQL instance address
• -p: port number (default: 5432)
• -U: subscriber, which must have the REPLICATION privilege
• -d: database name
• --slot: replication slot name
• --start: the position where the system begins consuming data. This parameter can be omitted for the first consumption. By default, the first consumption starts from the start position.
• -f: output to the console. The value - indicates standard output.

Example output:

BEGIN 1234
table public.tb_tests: INSERT: id[integer]:1 product_name[character varying]:'Laptop' price[numeric(10,2)]:5999.99 create_time[timestamp with time zone]:'2024-05-20 10:00:00+08'
table public.tb_tests: INSERT: id[integer]:2 product_name[character varying]:'Mechanical keyboard' price[numeric(10,2)]:299.99 create_time[timestamp with time zone]:'2024-05-20 10:01:00+08'
COMMIT 1234
BEGIN 1235
table public.tb_tests: UPDATE: id[integer]:1 price[numeric(10,2)]:5899.99 (old:5999.99)
COMMIT 1235
BEGIN 1236
table public.tb_tests: DELETE: id[integer]:2
COMMIT 1236

• After pg_recvlogical consumes data, the data is marked as processed and restart_lsn of the replication slot is updated to the latest position. pg_recvlogical and pg_logical_slot_peek_changes will not return these data (to prevent repeated consumption).
• The command remains connected. If there are new changes (for example, INSERT or UPDATE is executed again), new transaction data is generated in real time.
• To terminate the consumption, press Ctrl+C. The next execution will continue the consumption from the last position where it is terminated (depending on restart_lsn recorded by the replication slot).