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 assertCreateReplicationStreamSuccess() throws SQLException {
        LogSequenceNumber startPosition = LogSequenceNumber.valueOf(100L);
        when(connection.unwrap(PGConnection.class)).thenReturn(connection);
        when(connection.getReplicationAPI()).thenReturn(pgReplicationConnection);
        when(pgReplicationConnection.replicationStream()).thenReturn(chainedStreamBuilder);
        when(chainedStreamBuilder.logical()).thenReturn(chainedLogicalStreamBuilder);
        when(chainedLogicalStreamBuilder.withStartPosition(startPosition)).thenReturn(chainedLogicalStreamBuilder);
        when(chainedLogicalStreamBuilder.withSlotName("")).thenReturn(chainedLogicalStreamBuilder);
        when(chainedLogicalStreamBuilder.withSlotOption(anyString(), eq(true))).thenReturn(chainedLogicalStreamBuilder, chainedLogicalStreamBuilder);
        BaseLogSequenceNumber basePosition = new PostgreSQLLogSequenceNumber(startPosition);
        logicalReplication.createReplicationStream(connection, "", basePosition);
        verify(chainedLogicalStreamBuilder).start();
    }