private ServerInfo.ReplicationSlot fetchReplicationSlotInfo(String slotName, String pluginName)
            throws SQLException {
        final String database = database();
        final ServerInfo.ReplicationSlot slot =
                queryForSlot(
                        slotName,
                        database,
                        pluginName,
                        rs -> {
                            if (rs.next()) {
                                boolean active = rs.getBoolean("active");
                                final Lsn confirmedFlushedLsn =
                                        parseConfirmedFlushLsn(slotName, pluginName, database, rs);
                                if (confirmedFlushedLsn == null) {
                                    return null;
                                }
                                Lsn restartLsn =
                                        parseRestartLsn(slotName, pluginName, database, rs);
                                if (restartLsn == null) {
                                    return null;
                                }
                                final Long xmin = rs.getLong("catalog_xmin");
                                return new ServerInfo.ReplicationSlot(
                                        active, confirmedFlushedLsn, restartLsn, xmin);
                            } else {
                                LOGGER.debug(
                                        "No replication slot '{}' is present for plugin '{}' and database '{}'",
                                        slotName,
                                        pluginName,
                                        database);
                                return ServerInfo.ReplicationSlot.INVALID;
                            }
                        });
        return slot;
    }
Fetches the state of a replication stage given a slot name and plugin name

@param slotName the name of the slot
@param pluginName the name of the plugin used for the desired slot
@return the {@link ServerInfo.ReplicationSlot} object or a {@link
    ServerInfo.ReplicationSlot#INVALID} if the slot is not valid
@throws SQLException is thrown by the underlying JDBC
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/io/debezium/connector/postgresql/connection/PostgresConnection.java