Configuring Connection Details

This section provides instructions for configuring the connection details required to establish communication between the Micro-Integration and your third-party system.

For information about configuring the connection to the event broker, see Connecting to Your Event Broker.

This Micro-Integration supports workflows only in the direction of:

  • PostgreSQL to Solace

The name of the binder for PostgreSQL is pgcdc.

PostgreSQL Database Prerequisites

Before you use the Micro-Integration for PostgreSQL CDC, you must configure your PostgreSQL database with the following settings:

  • Logical replication enabled

    PostgreSQL must be configured with logical replication enabled. For more information, see the PostgreSQL documentation.

  • A publication for the table that is being monitored. A publication is a PostgreSQL feature that defines a set of tables whose changes are to be published.

    To create a publication for the table to monitor, run the following SQL command:

    CREATE PUBLICATION my_publication FOR TABLE my_table;

    You can set the publicationAutocreateMode property to automatically create the publication.

PostgreSQL Connection Details

To manually configure the PostgreSQL connection details, set the following properties in the application.yml file:

spring:
  cloud:
    stream:
      pgcdc:
        bindings:
          input-0:
            consumer:
              endpoint:
                query-parameters:
                  databaseHostname: <hostname>
                  databasePort: <port>
                  databaseDbname: <database name>
                  username: <database user/role name>
                  <authentication parameter>: <authentication parameter value>
                  <authentication parameter>: <authentication parameter value>
                  ...

Where:

  • databaseHostname is the hostname of the PostgreSQL database server.

  • databasePort is the connection port of the PostgreSQL database server.

  • databaseDbname is the name of the PostgreSQL database to connect to.

  • databaseUser is the PostgreSQL user to log in as. The PostgreSQL user determines the user role and permissions for the database.

  • <authentication parameter> is a parameter required for authenticating with the PostgreSQL database server. Choose one of:

Basic Authentication Parameters

Field Type Description
databaseUser String The username to use for logging in to PostgreSQL.
databasePassword String The password to use for authenticating with the PostgreSQL database.

Client Certificate Authentication Parameters

Field Type Description
databaseSslmode String

The SSL mode determines the security level for client-server communications. It controls whether SSL/TLS encryption is used and how rigorously the server's identity is verified. Valid values are:

  • disable to use an unencrypted connection

  • allow to try to use an unencrypted connection first and, failing that, a secure (encrypted) connection

  • prefer to try to use a secure (encrypted) connection first and, failing that, an unencrypted connection

  • require to use a secure (encrypted) connection, and fail if one cannot be established

  • verify-ca like required but additionally verify the server TLS certificate against the configured Certificate Authority (CA) certificates, or fail if no valid matching CA certificates are found

  • verify-full like verify-ca but additionally verify that the server certificate matches the host to which the connection is attempted

For more information, see the Protection Provided in Different Modes section of the PostgreSQL documentation.
databaseSslcert String The path to the SSL Certificate for the client, for example, local/ssl/client.crt.
databaseSslkey String The path to the file containing the SSL private key for the client.. for example, local/ssl/client.key
databaseSslpassword String The password to access the client private key from the file specified by databaseSslkey.
databaseSslrootcert String The path to the database SSL root certificate against which the server is validated, for example, local/ssl/root.crt.
databaseUser String The username to use for logging in to PostgreSQL.

Checkpoint Store Configuration

The Micro-Integration for PostgreSQL CDC stores the information about the current progress of processing files in a checkpoint store backed by a Last Value Queue (LVQ) on a Solace event broker. The checkpoint store contains information about the files that have been processed and the files that are currently being processed.

For example, add the following settings to the application.yml file:

solace:
  pgcdc:
    checkpoint:
      lvqName: lvq-checkpoint
      autoProvisionLvq: true

The following table lists the configuration options for the Checkpoint Store.

Config Option Type Valid Values Default Value Description

solace.pgcdc.checkpoint.lvqName

String

Required.

The name of the event broker Last Value Queue (spool size 0) to be used for checkpointing. The queue must exist on the same event broker and Message VPN as the target queue.

If the LVQ is deleted (administratively) or a message from the LVQ is deleted or consumed by another consumer, the Micro-Integration for PostgreSQL CDC will not be able to resume from the last checkpoint. In addition, the LVQ should not be shared by multiple instances of the Micro-Integration for PostgreSQL CDC.

solace.pgcdc.checkpoint.autoProvisionLvq

boolean

true, false

false

Optional.

Set to true to auto-provision the LVQ queue if it does not exist.

PostgreSQL CDC Binder Configuration Options

The following properties are available at the PostgreSQL CDC binder level.

Config Option Type Valid Values Description

spring.cloud.stream.bindings.input-<workflow-id>.destination

String

Format:

schema.tableName

For example, public.products

The PostgreSQL table from which to consume change events. The destination must include the schema and the name of the table.

spring.cloud.stream.bindings.input-<workflow-id>.binder

String

pgcdc

This property must be set to pgcdc.

PostgreSQL CDC Consumer Configuration Options

The following table lists the PostgreSQL CDC source-specific properties.

All properties must be prefixed with spring.cloud.stream.pgcdc.bindings.input-<workflow-id>.consumer.endpoint.query-parameters.

Example configuration:

pgcdc:
  bindings:
     input-0:
       consumer:
         endpoint:
           query-parameters:
             <connection parameters>
	      pluginName: pgoutput
             slotName: debezium
             tableIncludeList: public.inventory
             schemaIncludeList: public
             topicPrefix: cdc
             snapshotMode: initial
Config Option Type Valid Values Default Value Description

databaseDbName

String

The name of the PostgreSQL database.

slotName

String

The name of the PostgreSQL logical decoding slot. A logical decoding slot is a persistent object that maintains a stream of database changes from a plugin. The Micro-Integration for PostgreSQL CDC consumes the change stream from the specified slot. For more information, consult your PostgreSQL server administrator.

topicPrefix

String

The binding name (for example, input-0)

Topic prefix that identifies and provides a namespace for the particular database server/cluster that is capturing changes.

snapshotMode

String

no_data | always | initial | initial_only | never | custom

no_data

The criteria for running a snapshot upon startup of the Micro-Integration.

pluginName

String

pgoutput | decoderbufs

pgoutput

The name of the PostgreSQL logical decoding plugin installed on the server. The logical decoding plugin converts write-ahead log (WAL) entries into a stream of changes. For more information, consult your PostgreSQL server administrator.

databaseUser

String

The user to authenticate with PostgreSQL database.

databasePassword

String

Password to use for authentication.

databaseSslmode

String

The SSL mode to authenticate with the PostgreSQL database.

databaseSslcert

String

The SSL Certificate for the client.

databaseSslkey

String

The SSL private key for the client.

databaseSslpassword

String

The database SSL password to authenticate with the PostgreSQL database.

databaseSslrootcert

String

The database SSL root certificate against which the server is validated.

skippedOperations

String

t

A comma-separated list of operations to skip during streaming. Valid options include: c for inserts/create; u for updates; d for deletes, t for truncates, and none to indicate nothing skipped.

publicationName

String   dbz_publication The name of the PostgreSQL publication used for streaming changes from a plugin.

publicationAutocreateMode

String

disabled | all_tables | filtered

disabled

Specifies when a publication is to be created. Applies only when streaming changes using pgoutput. Select one of the following values:

  • disabled—The Micro-Integration will not attempt to create a publication at all. The expectation is that the user has created the publication up-front. If the publication isn’t found to exist upon startup, the Micro-Integration will throw an exception and stop.

  • all_tables—If no publication exists, the Micro-Integration will create a new publication for all tables:

    CREATE PUBLICATION FOR ALL TABLES;

    Note this requires that the configured user has access. If the publication already exists, it will be used.

  • filtered—If no publication exists, the Micro-Integration will create a new publication for all those tables matching the current filter configuration:

    CREATE PUBLICATION FOR TABLE;

    If the publication already exists, it will be used.

decimalHandlingMode

String precise | string | double precise

Specify how decimal and numeric columns should be represented in change events, including:

  • precise—Uses java.math.BigDecimal to represent values.

  • string—Uses string to represent values.

  • double—Represents values using Java’s double, which may not offer the precision but will be far easier to use in consumers.

timePrecisionMode

String adaptive | adaptive_time_microseconds | connect adaptive

Time, date, and timestamps can be represented with different kinds of precisions, including:

  • adaptive—Bases the precision of time, date, and timestamp values on the database column’s precision.

  • adaptive_time_microseconds—Like adaptive mode, but time fields always use microseconds precision.

  • connect—Uses millisecond precision regardless of the database columns' precision.

tombstonesOnDelete

boolean

false

Indicates whether delete operations should be represented by a delete event and a subsequent tombstone event.

skipDeleteEvents

boolean

true or false

true

Indicates whether to skip delete events. When true, this setting is equivalent to skippedOperations=t,d; when false, it is equivalent to skippedOperations=t.

tableIncludeList

This parameter is ignored.

schemaIncludeList

This parameter is ignored.

offsetStorageTopic

This parameter is ignored.

offsetStoragePartitions

This parameter is ignored.

offsetStorageReplicationFactor

This parameter is ignored.

PostgreSQL CDC Injected Headers

The following file-related headers are injected into each message. The examples illustrate the header values for an update to the PostgreSQL source table.

Header Name Type Description

cdc_operation

String

Values for the Micro-Integration are c for create (or insert), u for update, d for delete or r for read.

cdc_db_name

String

The name of the database.

cdc_schema

String

The name of the schema.

For example, public.

cdc_table_name

String

The name of the table.

For example, products.

cdc_lsn

Long

The log sequence number.

cdc_txId

Long

The transaction ID of the database transaction.

cdc_key

String

The key of the event.

For example, {"id": 5}.

cdc_before

String

The state of the row before the event occurred.

For example, {"id": 5, "key1": "value1", "key2": "value2"}.

cdc_timestamp

Long

The time at which the Micro-Integration processed the event.