PolarDB:Debezium connector for PolarDB for PostgreSQL (Compatible with Oracle)

Since PolarDB for PostgreSQL (Compatible with Oracle) and community PostgreSQL differ only in how they handle a few data types and built-in objects, this topic describes how to build a Debezium connector that supports PolarDB for PostgreSQL (Compatible with Oracle) by adapting the community's Debezium PostgreSQL connector with minimal code changes.

Build the Debezium PolarDBO connector

Usage

The Debezium PolarDBO connector reads incremental changes from a PolarDB for PostgreSQL (Compatible with Oracle) database using logical replication. The following conditions must be met before you can use the connector:

• The wal_level cluster parameter must be set to logical. This setting adds the information required for logical replication to the write-ahead logging (WAL).

  Note

  You can set the wal_level cluster parameter in the console. For more information, see Configure cluster parameters. Changing this parameter restarts the cluster. Plan this operation carefully according to your business needs.

• Run the ALTER TABLE schema.table REPLICA IDENTITY FULL; command to set the REPLICA IDENTITY of each subscribed table to FULL. This ensures that insert and update events contain the previous values for all columns, which guarantees data consistency.

  Note

  • REPLICA IDENTITY is a table-level setting specific to PostgreSQL. It determines whether the logical decoding plug-in includes the previous values of the involved table columns during INSERT and UPDATE events. For more information about the values for REPLICA IDENTITY, see REPLICA IDENTITY.

  • Setting the REPLICA IDENTITY of a subscribed table to FULL may require a table lock, which can affect your services. Plan this operation according to your business needs. You can run the following command to check if the current setting is FULL:

    SELECT relreplident = 'f' FROM pg_class WHERE relname = 'tablename';

• The max_wal_senders and max_replication_slots parameter values must be greater than the sum of replication slots currently in use and those required by your Kafka jobs.

• Use a privileged account or a standard account that has both LOGIN and REPLICATION permissions. The account must also have the SELECT permission on all subscribed tables for the initial snapshot.

• Connect only to the primary endpoint of the PolarDB cluster. Logical replication is not supported on the cluster endpoint.

• Set the connector.class parameter to io.debezium.connector.postgresql.PolarDBOConnector.

• We recommend that you set the plugin.name parameter to pgoutput. Otherwise, incremental parsing may produce garbled text for databases that do not use UTF-8 encoding. For more information, see the community documentation.

Example

The following example describes how to use the Debezium PolarDBO connector to synchronize the t1 and t2 tables from the dbz_db database in a PolarDB for PostgreSQL Oracle compatibility 2.0 cluster to a Kafka message queue.

Prerequisites

1. Set up Kafka

   1. Deploy a Kafka instance and ensure it is accessible from the Kafka Connect host. You can also use ApsaraMQ for Kafka. For more information, see Quick start.

   2. Create a topic named pg_dbz_event in the Kafka instance to receive messages.

      Note

      For testing, you can create a single-partition topic for easier viewing. For production environments, create a multi-partition topic.

2. Start Kafka Connect locally in distributed mode on port 8083.

   • Copy the Debezium PolarDBO connector JAR package to the plugin.path directory of Kafka Connect.

     # Replace ${plugin.path} with the actual path.
     mkdir ${plugin.path}/debezium-connector-polardbo
     cp debezium-connector-postgres-polardbo-v1.0-2.6.2.Final.jar ${plugin.path}/debezium-connector-polardbo

3. Set up PolarDB for PostgreSQL (Compatible with Oracle)

   1. On the PolarDB cluster purchase page, purchase a PolarDB for PostgreSQL (Compatible with Oracle) 2.0 cluster.

   2. Configure the PolarDB cluster to meet all the prerequisites listed in the Usage section.

   3. Create a privileged account. For more information, see Create an account.

   4. Obtain the cluster primary endpoint. For more information, see View connection endpoints. If the PolarDB cluster and the Kafka Connect instance are in the same availability zone, you can use the private endpoint. Otherwise, you must request a public endpoint. Add the address of the Kafka Connect instance to the PolarDB cluster whitelist. For more information, see Configure a cluster whitelist.

   5. In the console, create a database named dbz_db. For more information, see Create a database.

   6. Run the following statements to create the t1 and t2 tables in the dbz_db database and insert data.

      CREATE TABLE public.t1 (a int PRIMARY KEY, b text, c TIMESTAMP);
      ALTER TABLE public.t1 REPLICA IDENTITY FULL;
      INSERT INTO public.t1(a, b, c) VALUES(1, 'a', now());
      CREATE TABLE public.t2 (a int PRIMARY KEY, b text, c DATE);
      ALTER TABLE public.t2 REPLICA IDENTITY FULL;
      INSERT INTO public.t2(a, b, c) VALUES(1, 'a', now());

Test

1. Create a configuration file named config/postgresql-connector.json. For parameter descriptions, see the community documentation.

   {
     "name": "dbz-polardb",
     "config": {
       "connector.class": "io.debezium.connector.postgresql.PolarDBOConnector",
       "database.hostname": "<yourHostname>", 
       "database.port": "<yourPort>", 
       "database.user": "<yourUserName>", 
       "database.password": "<yourPassWord>", 
       "database.dbname" : "dbz_db",
       "plugin.name": "pgoutput",
       "slot.name": "dbz_polardb",
       "table.include.list": "public.t1,public.t2",
       "topic.prefix": "polardb"
       "transforms": "Combine",
       "transforms.Combine.type": "io.debezium.transforms.ByLogicalTableRouter",
       "transforms.Combine.topic.regex": "(.*)",
       "transforms.Combine.topic.replacement": "pg_dbz_event"
     }
   }

   Note

   By default, Debezium creates one topic per table. This configuration routes all change events to a single topic.

2. Add the connector.

   curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" 'http://localhost:8083/connectors' -d @config/postgresql-connector.json

   After the connector is added, the full data becomes available in the Kafka topic.

   On the message query tab of your Kafka instance, set the query method to query by offset, select partition 0, and set the start offset to 0. Then, click Query. The results show two Debezium CDC messages (at offsets 0 and 1). The keys contain __dbz__physicalTableIdentifier values of polardb.public.t1 and polardb.public.t2, respectively. This confirms that the full data snapshot has been synchronized to Kafka.

3. Run the following DML statements in the dbz_db database of the PolarDB cluster:

   INSERT INTO public.t1(a, b, c) VALUES(2, 'b', now());
   UPDATE public.t1 SET b = 'c' WHERE a = 1;
   DELETE FROM public.t1 WHERE a = 2;
   INSERT INTO public.t1(a, b, c) VALUES(4, 'd', now());
   INSERT INTO public.t2(a, b, c) VALUES(2, 'b', now());
   UPDATE public.t2 SET b = 'c' WHERE a = 1;
   DELETE FROM public.t2 WHERE a = 2;
   INSERT INTO public.t2(a, b, c) VALUES(4, 'd', now());

   The incremental data is now available in the Kafka topic.

   On the message query page, set the query method to query by offset, select the target partition and start offset, and then click Query. The results show that Debezium captured the DML operations on the t1 and t2 tables as CDC messages. The key of each message includes the table identifier (such as polardb.public.t1) and the primary key value. The message value contains the Debezium-formatted change event, including before and after fields. For a DELETE operation, the message value is 0 bytes.