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.

To share settings across all workflows in a Micro-Integration, configure the binder-specific parameters under default at the binder level instead of under a workflow-specific identifier (for example, input-0) in bindings. This reduces repetition when multiple workflows use the same parameter values. Workflow-specific settings override default settings when both are present.

Note that default is a sibling of bindings; it is not nested under bindings. Use the same hierarchy under default as you use under bindings.<binding-name>. For example, for bindings.input-0.consumer.someProperty=someValue the default setting is default.consumer.someProperty=someValue.

spring:
  cloud:
    stream:
      <binder-name>:
        default:
          <shared parameters>
        bindings:
          input-0:
            <workflow-specific parameters>

PostgreSQL Database Prerequisites

Before you use the Micro-Integration for PostgreSQL Bulk 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 tables that are 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 specific tables to monitor, run the following SQL command:

    CREATE PUBLICATION my_publication FOR TABLE table1, table2;

    Alternatively, to create a publication for all tables, run:

    CREATE PUBLICATION my_publication FOR ALL TABLES;

    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>
                  databaseUser: <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 Bulk CDC stores the information about the current progress of processing in a checkpoint store backed by a Last Value Queue (LVQ) on a Solace event broker. The checkpoint store contains information about the change events that have been processed and those 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 Bulk 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 Bulk CDC.

solace.pgcdc.checkpoint.autoProvisionLvq

boolean

true, false

false

Optional.

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

solace.pgcdc.checkpoint.cacheFlushFrequency

long

3000

Optional.

The frequency (in milliseconds) at which the checkpoint cache is flushed to the LVQ.

PostgreSQL Bulk CDC Binder Configuration Options

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

Config Option Type Valid Values Description

spring.cloud.stream.bindings.input-0.destination

String

Format:

schema.tableName

For multiple tables, use a semicolon-separated list:

schema.table1;schema.table2

Regular expressions are also supported, for example:

schema.table_.*

The PostgreSQL tables from which to consume change events. The destination must include the schema and the name of the table. You can specify multiple tables using a semicolon-separated list or a regular expression.

spring.cloud.stream.bindings.input-0.binder

String

pgcdc

This property must be set to pgcdc.

PostgreSQL Bulk CDC Consumer Configuration Options

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

All properties must be prefixed with spring.cloud.stream.pgcdc.bindings.input-0.consumer.endpoint.query-parameters.

Example configuration:

spring:
  cloud:
    stream:
      pgcdc:
        bindings:
          input-0:
            consumer:
              endpoint:
                query-parameters:
                  <connection parameters>
                  pluginName: pgoutput
                  slotName: debezium
                  tableIncludeList: public.inventory;public.products
                  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. Defaults to true.

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.