public PGReplicationStream createReplicationStream(final Connection connection, final String slotName, final BaseLogSequenceNumber startPosition) throws SQLException {
        return connection.unwrap(PGConnection.class).getReplicationAPI()
                .replicationStream()
                .logical()
                .withStartPosition((LogSequenceNumber) startPosition.get())
                .withSlotName(slotName)
                .withSlotOption("include-xids", true)
                .withSlotOption("skip-empty-xacts", true)
                .start();
    }
@Test
    void assertCreateReplicationStreamFailure() throws SQLException {
        when(connection.unwrap(PGConnection.class)).thenThrow(new SQLException(""));
        assertThrows(SQLException.class, () -> logicalReplication.createReplicationStream(connection, "", new PostgreSQLLogSequenceNumber(LogSequenceNumber.valueOf(100L))));
    }