> ## Documentation Index > Fetch the complete documentation index at: https://docs.streamnative.io/llms.txt > Use this file to discover all available pages before exploring further. > The Debezium source connector pulls messages from PostgreSQL and persists the messages to Pulsar topics. # Debezium postgres source The Postgres source connector pulls messages from PostgreSQL and persists the messages to Pulsar topics by using debezium. This connector is available as a built-in connector on StreamNative Cloud. ## Quick start ### Prerequisites The prerequisites for connecting a Debezium Postgres source connector to external systems include: 1. Create a Postgres service: This connector uses the debezium v3.2, Please refer to this [document](https://debezium.io/releases/3.2/) to see the compatible PostgreSQL versions. 2. Prepare Postgres Database: Please refer to this [document](https://debezium.io/documentation/reference/3.2/connectors/postgresql.html#setting-up-postgresql) to complete the prepare steps on Postgres. 3. Configure topic retention policies: Before running the connector, you must ensure that you have set an infinite retention policy for both the `offset.storage.topic` and `schema.history.internal.pulsar.topic`. Refer to the [Used Topic On Pulsar](#used-topic-on-pulsar) section for more details. The subsequent deployment steps detailed in this document leverage PostgreSQL 11.16 on AWS RDS, which natively supports the `pgoutput` plugin. ### 1. Create a table on Postgres Run the following SQL command on your PostgreSQL. If you don't require the `before` data, you can disregard the configuration of `REPLICA IDENTITY`. ```sql theme={null} CREATE TABLE "public"."io-test" ( "id" integer GENERATED ALWAYS AS IDENTITY, "first_name" text, "last_name" text, "age" integer, PRIMARY KEY ("id") ); ALTER TABLE "public"."io-test" REPLICA IDENTITY FULL; ``` ### 2. Create a connector The following command shows how to use [pulsarctl](https://github.com/streamnative/pulsarctl) to create a `builtin` connector. If you want to create a `non-builtin` connector, you need to replace `--source-type debezium-postgres` with `--archive /path/to/pulsar-io-debezium-postgres.nar`. You can find the button to download the `nar` package at the beginning of the document. If you are a StreamNative Cloud user, you need [set up your environment](https://docs.streamnative.io/docs/connector-setup) first. ```bash theme={null} pulsarctl sources create \ --source-type debezium-postgres \ --name debezium-postgres-source \ --tenant public \ --namespace default \ --parallelism 1 \ --source-config \ '{ "database.hostname": "Your hostname of Postgres", "database.port": "Your port of Postgres", "database.user": "Your user of Postgres", "database.password": "Your password of Postgres", "database.dbname": "Your dbname of Postgres", "table.include.list": "public.io-test", "database.server.name": "mydbserver", "topic.prefix": "postgres", "connector.class": "io.debezium.connector.postgresql.PostgresConnector", "plugin.name": "pgoutput" }' ``` 1. The `--parallelism` must be set to **1**. Debezium connectors do not support parallel consumption within a single instance. If you need to process tables in parallel, you can deploy multiple connector instances, each configured for different database schemas or tables. 2. You can set multiple tables for "table.include.list", and the connector will send data from each table to a different topic of pulsar. The topic naming rule is: `{{database.server.name}}.{{table.name}}`. For examples: `public/default/mydbserver.public.io-test`. The `--source-config` is the minimum necessary configuration for starting this connector, and it is a JSON string. You need to substitute the relevant parameters with your own. If you want to configure more parameters, see [Configuration Properties](#configuration-properties) for reference. You can also choose to use a variety of other tools to create a connector: * [pulsar-admin](https://pulsar.apache.org/docs/3.1.x/io-use/): The command arguments for `pulsar-admin` are similar to those of `pulsarctl`. You can find an example for [StreamNative Cloud Doc](https://docs.streamnative.io/docs/connector-create#create-a-built-in-connector). * [RestAPI](https://pulsar.apache.org/source-rest-api/?version=3.1.1): You can find an example for [StreamNative Cloud Doc](https://docs.streamnative.io/docs/connector-create#create-a-built-in-connector). * [Terraform](https://github.com/hashicorp/terraform): You can find an example for [StreamNative Cloud Doc](https://docs.streamnative.io/docs/connector-create#create-a-built-in-connector). * [Function Mesh](https://functionmesh.io/docs/connectors/run-connector): The docker image can be found at the beginning of the document. ### 3. Insert and update a data to table You can insert and update using the sql: ```sql theme={null} INSERT INTO "public"."io-test" (first_name, last_name, age) VALUES ('pg-io-test', 'streamnative', 4); UPDATE "public"."io-test" SET age = 5, last_name = 'sn' WHERE first_name = 'pg-io-test' AND last_name = 'streamnative'; ``` ### 4. Show data using Pulsar client If your connector is created on StreamNative Cloud, you need to authenticate your clients. See [Build applications using Pulsar clients](https://docs.streamnative.io/docs/qs-connect#jumpstart-for-beginners) for more information. ``` bin/pulsar-client \ --url "Your Pulsar serviceUrl" \ consume "public/default/mydbserver.public.io-test" -s "test-sub" -n 10 -p Earliest ----- got message ----- key:[eyJpZCI6Mn0=], properties:[], content:{"before":null,"after":{"id":1,"first_name":"pg-io-test","last_name":"streamnative","age":4},"source":{"version":"3.2.5.Final","connector":"postgresql","name":"mydbserver","ts_ms":1698825100079,"snapshot":"false","db":"postgres","sequence":"[null,\"18052284768\"]","schema":"public","table":"io-test","txId":2245,"lsn":18052284768,"xmin":null},"op":"c","ts_ms":1698825103451,"transaction":null} ----- got message ----- key:[eyJpZCI6M30=], properties:[], content:{"before":{"id":1,"first_name":"pg-io-test","last_name":"streamnative","age":4},"after":{"id":1,"first_name":"pg-io-test","last_name":"sn","age":5},"source":{"version":"3.2.5.Final","connector":"postgresql","name":"mydbserver","ts_ms":1698826703631,"snapshot":"false","db":"postgres","sequence":"[\"18387831504\",\"18387832144\"]","schema":"public","table":"io-test","txId":2284,"lsn":18387832144,"xmin":null},"op":"u","ts_ms":1698826704159,"transaction":null} ``` ## Configuration Properties The configuration of Debezium source connector has the following properties. | Name | Required | Sensitive | Default | Description | | -------------------------------------------- | -------- | --------- | ------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `database.hostname` | true | false | null | The address of a database server. | | `database.port` | true | false | null | The port number of a database server. | | `database.user` | true | true | null | The name of a database user that has the required privileges. | | `database.password` | true | true | null | The password for a database user that has the required privileges. | | `database.dbname` | true | false | null | The database.dbname parameter in Debezium configuration is used to specify the name of the specific database that the connector should connect to. | | `plugin.name` | true | false | null | The plugin.name parameter in Debezium configuration is used to specify the logical decoding output plugin installed on the PostgreSQL server that the connector should use: `decoderbufs`, `wal2json`, `pgoutput` | | `database.server.name` | true | false | null | The logical name of a database server/cluster, which forms a namespace and it is used in all the names of Kafka topics to which the connector writes, the Kafka Connect schema names, and the namespaces of the corresponding Avro schema when the Avro Connector is used. | | `topic.prefix` | true | false | null | The prefix that is used to name the Kafka topics to which the connector writes the change events. | | `connector.class` | true | false | null | The name of the Debezium connector class to use. Can only be: `io.debezium.connector.postgresql.PostgresConnector`. | | `database.server.id` | false | false | null | The connector’s identifier that must be unique within a database cluster and similar to the database’s server-id configuration property. | | `schema.include.list` | false | false | null | A list of all schemas hosted by this server which is monitored by the connector.

This is optional, and there are other properties for listing schemas and tables to include or exclude from monitoring. | | `schema.exclude.list` | false | false | null | A list of all schemas hosted by this server which is not monitored by the connector.

This is optional, and there are other properties for listing schemas and tables to include or exclude from monitoring. | | `table.include.list` | false | false | null | A list of all tables hosted by this server which is monitored by the connector.

This is optional, and there are other properties for listing tables and tables to include or exclude from monitoring. | | `table.exclude.list` | false | false | null | A list of all tables hosted by this server which is not monitored by the connector.

This is optional, and there are other properties for listing tables and tables to include or exclude from monitoring. | | `key.converter` | false | false | null | The converter provided by Kafka Connect to convert record key. | | `value.converter` | false | false | null | The converter provided by Kafka Connect to convert record value. | | `schema.history.internal` | false | false | null | The name of the database history class. | | `schema.history.internal.pulsar.topic` | false | false | null | The name of the database history topic where the connector writes and recovers DDL statements.

**Note: this topic is for internal use only and should not be used by consumers.** | | `schema.history.internal.pulsar.service.url` | false | false | null | Pulsar cluster service URL for history topic. | | `pulsar.service.url` | false | false | null | Pulsar cluster service URL. | | `offset.storage.topic` | false | false | null | Record the last committed offsets that the connector successfully completes. | For more configuration properties, please see [Debezium PostgreSQL connector configuration properties](https://debezium.io/documentation/reference/3.2/connectors/postgresql.html#postgresql-connector-properties) ## Advanced features ### Converter options * org.apache.kafka.connect.json.JsonConverter The`json-with-envelope` config is valid only for the JsonConverter. By default, the value is set to false. When the `json-with-envelope` value is set to false, the consumer uses the schema `Schema.KeyValue(Schema.AUTO_CONSUME(), Schema.AUTO_CONSUME(), KeyValueEncodingType.SEPARATED)`, and the message only consists of the payload. When the `json-with-envelope` value is set to true, the consumer uses the schema `Schema.KeyValue(Schema.BYTES, Schema.BYTES)`, and the message consists of the schema and the payload. * org.apache.pulsar.kafka.shade.io.confluent.connect.avro.AvroConverter If you select the AvroConverter, the consumer uses the schema `Schema.KeyValue(Schema.AUTO_CONSUME(), Schema.AUTO_CONSUME(), KeyValueEncodingType.SEPARATED)`, and the message consists of the payload. ### Used topic on Pulsar Currently, the destination topic (specified by the `destination-topic-name` option ) is a required configuration but it is not used for the Debezium connector to save data. The Debezium connector saves data on the following 4 types of topics: * One topic for storing the database metadata messages. It is named with the database server name ( `database.server.name`), like `public/default/database.server.name`. * One topic (`offset.storage.topic`) for storing the offset metadata messages. The connector saves the last successfully-committed offsets on this topic. * One topic (`schema.history.internal.pulsar.topic`) for storing the database history information. The connector writes and recovers DDL statements on this topic. * One per-table topic. You can set multiple tables for "table.include.list", and the connector will send data from each table to a different topic of pulsar. The topic naming rule is: `{{database.server.name}}.{{table.name}}`. For examples: `public/default/mydbserver.public.io-test`. If automatic topic creation is disabled on the Pulsar broker, you need to manually create these 4 types of topics and the destination topic. For `offset.storage.topic` and `schema.history.internal.pulsar.topic`, If they are not specified in your connector's configuration, they will be created automatically using the following default naming convention: * `schema.history.internal.pulsar.topic`: `"{tenant}/{namespace}/{connector-name}-debezium-history-topic"` * `offset.storage.topic`: `"{tenant}/{namespace}/{connector-name}-offset-storage-topic"` Here, {tenant} and {namespace} refer to the tenant and namespace where the connector is running. Both the history and offset topics require their data to be retained indefinitely to ensure fault-tolerance and prevent data loss. Before running the connector, you must configure an infinite retention policy for both topics. Use the pulsar-admin CLI to set the retention policy: ```shell theme={null} pulsar-admin topicPolicies set-retention -s -1 -t -1 ${topic_name} ```