PostgreSQL CDC using Kafka and Debezium

As a superuser:

1. Check the wal_level configuration setting:

   SHOW wal_level;
   

   The default value is replica. For CDC, you’ll need to set it to logical in the database configuration file (postgresql.conf). Keep in mind that changing the wal_level requires a restart of the Postgres instance and can affect database performance.

2. Restart the database so all changes can take effect.

We recommend following the AWS RDS documentation for detailed information on logical replication configuration and best practices.

As a superuser (rds_superuser):

1. Create a custom RDS parameter group and associate it with your instance. You will not be able to set custom parameters on the default RDS parameter groups.

2. In the custom RDS parameter group, set the rds.logical_replication static parameter to 1.

3. Add the egress IP addresses associated with your Materialize region to the security group of the RDS instance. You can find these addresses by querying the mz_egress_ips table in Materialize.

4. Restart the database so all changes can take effect.

We recommend following the AWS Aurora documentation for detailed information on logical replication configuration and best practices.

As a superuser:

1. Create a DB cluster parameter group for your instance using the following settings:

   Set Parameter group family to your version of Aurora PostgreSQL.

   Set Type to DB Cluster Parameter Group.

2. In the DB cluster parameter group, set the rds.logical_replication static parameter to 1.

3. In the DB cluster parameter group, set reasonable values for max_replication_slots, max_wal_senders, max_logical_replication_workers, and max_worker_processes parameters based on your expected usage.

4. Add the egress IP addresses associated with your Materialize region to the security group of the DB instance. You can find these addresses by querying the mz_egress_ips table in Materialize.

5. Restart the database so all changes can take effect.

We recommend following the Azure DB for PostgreSQL documentation for detailed information on logical replication configuration and best practices.

1. In the Azure portal, or using the Azure CLI, enable logical replication for the PostgreSQL instance.

2. Add the egress IP addresses associated with your Materialize region to the list of allowed IP addresses under the “Connections security” menu. You can find these addresses by querying the mz_egress_ips table in Materialize.

3. Restart the database so all changes can take effect.

We recommend following the Cloud SQL for PostgreSQL documentation for detailed information on logical replication configuration and best practices.

As a superuser (cloudsqlsuperuser):

1. In the Google Cloud Console, enable logical replication by setting the cloudsql.logical_decoding configuration parameter to on.

2. Add the egress IP addresses associated with your Materialize region to the list of allowed IP addresses. You can find these addresses by querying the mz_egress_ips table in Materialize.

3. Restart the database so all changes can take effect.

1. Create a connector configuration file and save it as register-postgres.json:

   {
       "name": "your-connector",
       "config": {
           "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
           "tasks.max": "1",
           "plugin.name":"pgoutput",
           "database.hostname": "postgres",
           "database.port": "5432",
           "database.user": "postgres",
           "database.password": "postgres",
           "database.dbname" : "postgres",
           "database.server.name": "pg_repl",
           "table.include.list": "public.table1",
           "publication.autocreate.mode":"filtered",
           "key.converter": "io.confluent.connect.avro.AvroConverter",
           "value.converter": "io.confluent.connect.avro.AvroConverter",
           "value.converter.schemas.enable": false
       }
   }
   

   You can read more about each configuration property in the Debezium documentation. By default, the connector writes events for each table to a Kafka topic named serverName.schemaName.tableName.

1. Beginning with Debezium 2.0.0, Confluent Schema Registry support is not included in the Debezium containers. To enable the Confluent Schema Registry for a Debezium container, install the following Confluent Avro converter JAR files into the Connect plugin directory:

   • kafka-connect-avro-converter
   • kafka-connect-avro-data
   • kafka-avro-serializer
   • kafka-schema-serializer
   • kafka-schema-registry-client
   • common-config
   • common-utils

   You can read more about this in the Debezium documentation.

2. Create a connector configuration file and save it as register-postgres.json:

   {
       "name": "your-connector",
       "config": {
           "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
           "tasks.max": "1",
           "plugin.name":"pgoutput",
           "database.hostname": "postgres",
           "database.port": "5432",
           "database.user": "postgres",
           "database.password": "postgres",
           "database.dbname" : "postgres",
           "topic.prefix": "pg_repl",
           "schema.include.list": "public",
           "table.include.list": "public.table1",
           "publication.autocreate.mode":"filtered",
           "key.converter": "io.confluent.connect.avro.AvroConverter",
           "value.converter": "io.confluent.connect.avro.AvroConverter",
           "key.converter.schema.registry.url": "http://<scheme-registry>:8081",
           "value.converter.schema.registry.url": "http://<scheme-registry>:8081",
           "value.converter.schemas.enable": false
       }
   }
   

   You can read more about each configuration property in the Debezium documentation. By default, the connector writes events for each table to a Kafka topic named serverName.schemaName.tableName.