PostgreSQLWalDumper extends AbstractShardingScalingExecutor<WalPosition> implements LogDumper { @Override public void start() { super.start(); dump(); } PostgreSQLWalDumper(final DumperConfiguration dumperConfig, final Position position); @Override void start();  }
@Test public void assertStart() throws SQLException, NoSuchFieldException, IllegalAccessException { try { ReflectionUtil.setFieldValueToClass(postgreSQLWalDumper, "logicalReplication", logicalReplication); when(logicalReplication.createPgConnection(jdbcDataSourceConfig)).thenReturn(pgConnection); when(pgConnection.unwrap(PgConnection.class)).thenReturn(pgConnection); when(pgConnection.getTimestampUtils()).thenReturn(null); when(logicalReplication.createReplicationStream(pgConnection, PostgreSQLPositionManager.SLOT_NAME, position.getLogSequenceNumber())).thenReturn(pgReplicationStream); ByteBuffer data = ByteBuffer.wrap("table public.test: DELETE: data[integer]:1".getBytes()); when(pgReplicationStream.readPending()).thenReturn(null).thenReturn(data).thenThrow(new SQLException("")); when(pgReplicationStream.getLastReceiveLSN()).thenReturn(LogSequenceNumber.valueOf(101L)); postgreSQLWalDumper.start(); } catch (final SyncTaskExecuteException ignore) { } assertThat(channel.fetchRecords(100, 0).size(), is(1)); }