com.hazelcast.jet.cdc.postgres

Class PostgresCdcSources.Builder

java.lang.Object

com.hazelcast.jet.cdc.postgres.PostgresCdcSources.Builder

Enclosing class:

PostgresCdcSources

public static final class PostgresCdcSources.Builder
extends Object

Builder for configuring a CDC source that streams change data from a PostgreSQL database to Hazelcast Jet.

Method Summary

Modifier and Type	Method and Description
StreamSource<ChangeRecord>	build() Returns the source based on the properties set so far.
PostgresCdcSources.Builder	setColumnBlacklist(String... columnNameRegExps) Optional regular expressions that match the fully-qualified names of columns that should be excluded from change event message values.
PostgresCdcSources.Builder	setCommitPeriod(long milliseconds) Specifies how often the connector should confirm processed offsets to the Postgres database's replication slot.
PostgresCdcSources.Builder	setCustomProperty(String key, String value) Can be used to set any property not explicitly covered by other methods or to override internal properties.
PostgresCdcSources.Builder	setCustomSnapshotter(Class<?> snapshotterClass) Custom snapshotter that will be used by the connector.
PostgresCdcSources.Builder	setDatabaseAddress(String address) IP address or hostname of the database server, has to be specified.
PostgresCdcSources.Builder	setDatabaseName(String dbName) The name of the PostgreSQL database from which to stream the changes.
PostgresCdcSources.Builder	setDatabasePassword(String password) Database user password for connecting to the database server.
PostgresCdcSources.Builder	setDatabasePort(int port) Optional port number of the database server, if unspecified defaults to the database specific default port (5432).
PostgresCdcSources.Builder	setDatabaseUser(String user) Database user for connecting to the database server.
PostgresCdcSources.Builder	setLogicalDecodingPlugIn(String pluginName) The name of the @see Postgres logical decoding plug-in installed on the server.
PostgresCdcSources.Builder	setPublicationName(String publicationName) The name of the Postgres publication that will be used for CDC purposes.
PostgresCdcSources.Builder	setReconnectBehavior(RetryStrategy retryStrategy) Specifies how the connector should behave when it detects that the backing database has been shut dow.
PostgresCdcSources.Builder	setReplicationSlotDropOnStop(boolean dropOnStop) Whether or not to drop the logical replication slot when the connector disconnects cleanly.
PostgresCdcSources.Builder	setReplicationSlotName(String slotName) The name of the @see Postgres logical decoding slot (also called "replication slot") created for streaming changes from a plug-in and database instance.
PostgresCdcSources.Builder	setSchemaBlacklist(String... schemaNameRegExps) Optional regular expressions that match schema names to be excluded from monitoring ("schema" is used here to denote logical groups of tables).
PostgresCdcSources.Builder	setSchemaWhitelist(String... schemaNameRegExps) Optional regular expressions that match schema names to be monitored ("schema" is used here to denote logical groups of tables).
PostgresCdcSources.Builder	setShouldStateBeResetOnReconnect(boolean reset) Specifies if the source's state should be kept or discarded during reconnect attempts to the database.
PostgresCdcSources.Builder	setSnapshotMode(PostgresCdcSources.PostgresSnapshotMode snapshotMode) Snapshot mode that will be used by the connector.
PostgresCdcSources.Builder	setSslCertificateFile(String file) Specifies the (path to the) file containing the SSL Certificate for the database client.
PostgresCdcSources.Builder	setSslKeyFile(String file) Specifies the (path to the) file containing the SSL private key of the database client.
PostgresCdcSources.Builder	setSslKeyFilePassword(String password) Specifies the password to be used to access the SSL key file, if specified.
PostgresCdcSources.Builder	setSslMode(String mode) Specifies whether to use an encrypted connection to the database.
PostgresCdcSources.Builder	setSslRootCertificateFile(String file) Specifies the file containing containing SSL certificate authority (CA) certificate(s).
PostgresCdcSources.Builder	setTableBlacklist(String... tableNameRegExps) Optional regular expressions that match fully-qualified table identifiers for tables to be excluded from monitoring; any table not included in the blacklist will be monitored.
PostgresCdcSources.Builder	setTableWhitelist(String... tableNameRegExps) Optional regular expressions that match fully-qualified table identifiers for tables to be monitored; any table not included in the whitelist will be excluded from monitoring.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Method Detail

setSnapshotMode

@Nonnull
public PostgresCdcSources.Builder setSnapshotMode(@Nonnull
                                                           PostgresCdcSources.PostgresSnapshotMode snapshotMode)

Snapshot mode that will be used by the connector. If you want to use PostgresConnectorConfig.SnapshotMode.CUSTOM, please use setCustomSnapshotter(Class) method instead.

setCustomSnapshotter

@Nonnull
public PostgresCdcSources.Builder setCustomSnapshotter(@Nonnull
                                                                Class<?> snapshotterClass)

Custom snapshotter that will be used by the connector.

setDatabaseAddress

@Nonnull
public PostgresCdcSources.Builder setDatabaseAddress(@Nonnull
                                                              String address)

IP address or hostname of the database server, has to be specified.

setDatabasePort

@Nonnull
public PostgresCdcSources.Builder setDatabasePort(int port)

Optional port number of the database server, if unspecified defaults to the database specific default port (5432).

setDatabaseUser

@Nonnull
public PostgresCdcSources.Builder setDatabaseUser(@Nonnull
                                                           String user)

Database user for connecting to the database server. Has to be specified.

setDatabasePassword

@Nonnull
public PostgresCdcSources.Builder setDatabasePassword(@Nonnull
                                                               String password)

Database user password for connecting to the database server. Has to be specified.

setDatabaseName

@Nonnull
public PostgresCdcSources.Builder setDatabaseName(@Nonnull
                                                           String dbName)

The name of the PostgreSQL database from which to stream the changes. Has to be set.

Currently this source is not capable of monitoring multiple databases, only multiple schemas and/or tables. See white- and black-listing configuration options for those.

setSchemaWhitelist

@Nonnull
public PostgresCdcSources.Builder setSchemaWhitelist(@Nonnull
                                                              String... schemaNameRegExps)

Optional regular expressions that match schema names to be monitored ("schema" is used here to denote logical groups of tables). Any schema name not included in the whitelist will be excluded from monitoring. By default all non-system schemas will be monitored. May not be used with schema blacklist.

setSchemaBlacklist

@Nonnull
public PostgresCdcSources.Builder setSchemaBlacklist(@Nonnull
                                                              String... schemaNameRegExps)

Optional regular expressions that match schema names to be excluded from monitoring ("schema" is used here to denote logical groups of tables). Any schema name not included in the blacklist will be monitored, with the exception of system schemas. May not be used with schema whitelist.

setTableWhitelist

@Nonnull
public PostgresCdcSources.Builder setTableWhitelist(@Nonnull
                                                             String... tableNameRegExps)

Optional regular expressions that match fully-qualified table identifiers for tables to be monitored; any table not included in the whitelist will be excluded from monitoring. Each identifier is of the form schemaName.tableName. By default the connector will monitor every non-system table in each monitored database. May not be used with table blacklist.

setTableBlacklist

@Nonnull
public PostgresCdcSources.Builder setTableBlacklist(@Nonnull
                                                             String... tableNameRegExps)

Optional regular expressions that match fully-qualified table identifiers for tables to be excluded from monitoring; any table not included in the blacklist will be monitored. Each identifier is of the form schemaName.tableName. May not be used with table whitelist.

setColumnBlacklist

@Nonnull
public PostgresCdcSources.Builder setColumnBlacklist(@Nonnull
                                                              String... columnNameRegExps)

Optional regular expressions that match the fully-qualified names of columns that should be excluded from change event message values. Fully-qualified names for columns are of the form schemaName.tableName.columnName.

setLogicalDecodingPlugIn

@Nonnull
public PostgresCdcSources.Builder setLogicalDecodingPlugIn(@Nonnull
                                                                    String pluginName)

The name of the @see Postgres logical decoding plug-in installed on the server. Supported values are decoderbufs, wal2json, wal2json_rds, wal2json_streaming, wal2json_rds_streaming and pgoutput.

If not explicitly set, the property defaults to decoderbufs.

When the processed transactions are very large it is possible that the JSON batch event with all changes in the transaction will not fit into the hard-coded memory buffer of size 1 GB. In such cases it is possible to switch to so-called streaming mode when every change in transactions is sent as a separate message from PostgreSQL.

setReplicationSlotName

@Nonnull
public PostgresCdcSources.Builder setReplicationSlotName(@Nonnull
                                                                  String slotName)

The name of the @see Postgres logical decoding slot (also called "replication slot") created for streaming changes from a plug-in and database instance.

Values must conform to Postgres replication slot naming rules which state: "Each replication slot has a name, which can contain lower-case letters, numbers, and the underscore character."

Replication slots have to have an identifier that is unique across all databases in a PostgreSQL cluster.

If not explicitly set, the property defaults to debezium.

setReplicationSlotDropOnStop

@Nonnull
public PostgresCdcSources.Builder setReplicationSlotDropOnStop(boolean dropOnStop)

Whether or not to drop the logical replication slot when the connector disconnects cleanly.

Defaults to false

Should only be set to true in testing or development environments. Dropping the slot allows WAL segments to be discarded by the database, so it may happen that after a restart the connector cannot resume from the WAL position where it left off before.

setPublicationName

@Nonnull
public PostgresCdcSources.Builder setPublicationName(@Nonnull
                                                              String publicationName)

The name of the Postgres publication that will be used for CDC purposes.

If the publication does not exist when this source starts up, then the source will create it (note: the database user of the source must have superuser permissions to be able to do so). If created this way the publication will include all tables and the source itself must filter the data based on its white-/blacklist configs. This is not efficient because the database will still send all data to the connector, before filtering is applied.

It's best to use a pre-defined publication (via the CREATE PUBLICATION SQL command, specified via its name.

If not explicitly set, the property defaults to dbz_publication.

setSslMode

@Nonnull
public PostgresCdcSources.Builder setSslMode(@Nonnull
                                                      String mode)

Specifies whether to use an encrypted connection to the database. The default is disable, and specifies to use an unencrypted connection.

The require option establishes an encrypted connection but will fail if one cannot be made for any reason.

The verify_ca option behaves like require but additionally it verifies the server TLS certificate against the configured Certificate Authority (CA) certificates and will fail if it doesn’t match any valid CA certificates.

The verify-full option behaves like verify_ca but additionally verifies that the server certificate matches the host of the remote connection.

setSslCertificateFile

@Nonnull
public PostgresCdcSources.Builder setSslCertificateFile(@Nonnull
                                                                 String file)

Specifies the (path to the) file containing the SSL Certificate for the database client.

setSslKeyFile

@Nonnull
public PostgresCdcSources.Builder setSslKeyFile(@Nonnull
                                                         String file)

Specifies the (path to the) file containing the SSL private key of the database client.

setSslKeyFilePassword

@Nonnull
public PostgresCdcSources.Builder setSslKeyFilePassword(@Nonnull
                                                                 String password)

Specifies the password to be used to access the SSL key file, if specified.

Mandatory if key file specified.

setSslRootCertificateFile

@Nonnull
public PostgresCdcSources.Builder setSslRootCertificateFile(@Nonnull
                                                                     String file)

Specifies the file containing containing SSL certificate authority (CA) certificate(s).

setReconnectBehavior

@Nonnull
public PostgresCdcSources.Builder setReconnectBehavior(@Nonnull
                                                                RetryStrategy retryStrategy)

Specifies how the connector should behave when it detects that the backing database has been shut dow.

Defaults to CdcSourceP.DEFAULT_RECONNECT_BEHAVIOR.

setShouldStateBeResetOnReconnect

@Nonnull
public PostgresCdcSources.Builder setShouldStateBeResetOnReconnect(boolean reset)

Specifies if the source's state should be kept or discarded during reconnect attempts to the database. If the state is kept, then database snapshotting should not be repeated and streaming the binlog should resume at the position where it left off. If the state is reset, then the source will behave as if it were its initial start, so will do a database snapshot and will start tailing the binlog where it syncs with the database snapshot's end.

setCommitPeriod

@Nonnull
public PostgresCdcSources.Builder setCommitPeriod(long milliseconds)

Specifies how often the connector should confirm processed offsets to the Postgres database's replication slot. For jobs with a processing guarantee this option is ignored, the source confirms the offsets after each state snapshot.

If set to zero, the connector will commit the offsets after each batch of change records.

If set to a positive value, the commits will be done in the given period.

Negative values are not allowed.

Defaults to CdcSourceP.DEFAULT_COMMIT_PERIOD_MS.

Since:

Jet 4.4.1

setCustomProperty

@Nonnull
public PostgresCdcSources.Builder setCustomProperty(@Nonnull
                                                             String key,
                                                             @Nonnull
                                                             String value)

Can be used to set any property not explicitly covered by other methods or to override internal properties.

build

@Nonnull
public StreamSource<ChangeRecord> build()

Returns the source based on the properties set so far.