@VisibleForTesting
    ServerInfo.ReplicationSlot readReplicationSlotInfo(String slotName, String pluginName)
            throws SQLException, InterruptedException {
        final String database = database();
        final Metronome metronome =
                Metronome.parker(PAUSE_BETWEEN_REPLICATION_SLOT_RETRIEVAL_ATTEMPTS, Clock.SYSTEM);

        for (int attempt = 1; attempt <= MAX_ATTEMPTS_FOR_OBTAINING_REPLICATION_SLOT; attempt++) {
            final ServerInfo.ReplicationSlot slot = fetchReplicationSlotInfo(slotName, pluginName);
            if (slot != null) {
                LOGGER.info("Obtained valid replication slot {}", slot);
                return slot;
            }
            LOGGER.warn(
                    "Cannot obtain valid replication slot '{}' for plugin '{}' and database '{}' [during attempt {} out of {}, concurrent tx probably blocks taking snapshot.",
                    slotName,
                    pluginName,
                    database,
                    attempt,
                    MAX_ATTEMPTS_FOR_OBTAINING_REPLICATION_SLOT);
            metronome.pause();
        }

        throw new ConnectException(
                "Unable to obtain valid replication slot. "
                        + "Make sure there are no long-running transactions running in parallel as they may hinder the allocation of the replication slot when starting this connector");
    }
Fetches a replication slot, repeating the query until either the slot is created or until the
max number of attempts has been reached

<p>To fetch the slot without the retries, use the {@link
PostgresConnection#fetchReplicationSlotInfo} call

@param slotName the slot name
@param pluginName the name of the plugin
@return the {@link ServerInfo.ReplicationSlot} object or a {@link
    ServerInfo.ReplicationSlot#INVALID} if the slot is not valid
@throws SQLException is thrown by the underyling jdbc driver
@throws InterruptedException is thrown if we don't return an answer within the set number of
    retries
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/io/debezium/connector/postgresql/connection/PostgresConnection.java