public static ReplicationConnection createReplicationConnection(
            PostgresTaskContext taskContext,
            boolean doSnapshot,
            PostgresConnectorConfig connectorConfig) {
        int maxRetries = connectorConfig.maxRetries();
        Duration retryDelay = connectorConfig.retryDelay();

        final Metronome metronome = Metronome.parker(retryDelay, Clock.SYSTEM);
        short retryCount = 0;
        while (retryCount <= maxRetries) {
            try {
                LOGGER.info("Creating a new replication connection for {}", taskContext);
                return taskContext.createReplicationConnection(doSnapshot);
            } catch (SQLException ex) {
                retryCount++;
                if (retryCount > maxRetries) {
                    LOGGER.error(
                            "Too many errors connecting to server. All {} retries failed.",
                            maxRetries);
                    throw new FlinkRuntimeException(ex);
                }

                LOGGER.warn(
                        "Error connecting to server; will attempt retry {} of {} after {} "
                                + "seconds. Exception message: {}",
                        retryCount,
                        maxRetries,
                        retryDelay.getSeconds(),
                        ex.getMessage());
                try {
                    metronome.pause();
                } catch (InterruptedException e) {
                    LOGGER.warn("Connection retry sleep interrupted by exception: " + e);
                    Thread.currentThread().interrupt();
                }
            }
        }
        LOGGER.error("Failed to create replication connection after {} retries", maxRetries);
        throw new FlinkRuntimeException(
                "Failed to create replication connection for " + taskContext);
    }
Create a new PostgresVauleConverterBuilder instance and offer type registry for JDBC
connection.

<p>It is created in this package because some methods (e.g., includeUnknownDatatypes) of
PostgresConnectorConfig is protected.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/io/debezium/connector/postgresql/PostgresObjectFactory.java