@Override
  public UnboundedReader<KV<byte[], byte[]>> createReader(
      PipelineOptions options, @Nullable SyntheticRecordsCheckpoint checkpoint) {

    if (checkpoint == null) {
      return new SyntheticUnboundedReader(this, this.startOffset);
    } else {
      return new SyntheticUnboundedReader(this, checkpoint.getCurrentCheckMarkPosition());
    }
  }
@Test
  public void shouldStartTheReaderSuccessfully() throws IOException {
    UnboundedSource.UnboundedReader<KV<byte[], byte[]>> reader =
        source.createReader(pipeline.getOptions(), checkpoint);

    boolean isStarted = reader.start();
    assertTrue(isStarted);
  }