LogicalReplication { public PGReplicationStream createReplicationStream(final Connection pgConnection, final String slotName, final LogSequenceNumber startPosition) throws SQLException { return pgConnection.unwrap(PGConnection.class).getReplicationAPI() .replicationStream() .logical() .withStartPosition(startPosition) .withSlotName(slotName) .withSlotOption("include-xids", true) .withSlotOption("skip-empty-xacts", true) .start(); }  Connection createPgConnection(final JDBCScalingDataSourceConfiguration jdbcDataSourceConfig); PGReplicationStream createReplicationStream(final Connection pgConnection, final String slotName, final LogSequenceNumber startPosition);  }
@Test public void assertCreateReplicationStreamSuccess() throws SQLException { LogSequenceNumber startPosition = LogSequenceNumber.valueOf(100L); when(pgConnection.unwrap(PGConnection.class)).thenReturn(pgConnection); when(pgConnection.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); logicalReplication.createReplicationStream(pgConnection, "", startPosition); verify(chainedLogicalStreamBuilder).start(); }
@Test(expected = SQLException.class) @SneakyThrows(SQLException.class) public void assertCreateReplicationStreamFailure() { when(pgConnection.unwrap(PGConnection.class)).thenThrow(new SQLException("")); logicalReplication.createReplicationStream(pgConnection, "", LogSequenceNumber.valueOf(100L)); }