SyntheticUnboundedSource extends UnboundedSource<KV<byte[], byte[]>, SyntheticRecordsCheckpoint> { @Override public UnboundedReader<KV<byte[], byte[]>> createReader( PipelineOptions options, @Nullable SyntheticRecordsCheckpoint checkpoint) { if (checkpoint == null) { return new SyntheticUnboundedReader(this); } else { return new SyntheticUnboundedReader( new SyntheticUnboundedSource( checkpoint.getStartPosition(), checkpoint.getEndPosition(), sourceOptions)); } } SyntheticUnboundedSource(SyntheticSourceOptions sourceOptions); private  SyntheticUnboundedSource(
      long startOffset, long endOffset, SyntheticSourceOptions sourceOptions); @Override Coder<KV<byte[], byte[]>> getOutputCoder(); @Override void validate(); @Override Coder<SyntheticRecordsCheckpoint> getCheckpointMarkCoder(); @Override UnboundedReader<KV<byte[], byte[]>> createReader(
      PipelineOptions options, @Nullable SyntheticRecordsCheckpoint checkpoint); @Override List<SyntheticUnboundedSource> split(int desiredNumSplits, PipelineOptions options);  }
@Test public void shouldStartTheReaderSuccessfully() throws IOException { UnboundedSource.UnboundedReader<KV<byte[], byte[]>> reader = source.createReader(pipeline.getOptions(), checkpoint); boolean isStarted = reader.start(); assertTrue(isStarted); }
@Test public void startPositionShouldBeExclusive() throws IOException { int startPosition = 0; checkpoint = new SyntheticRecordsCheckpoint(startPosition, sourceOptions.numRecords); UnboundedSource.UnboundedReader<KV<byte[], byte[]>> reader = source.createReader(pipeline.getOptions(), checkpoint); reader.start(); KV<byte[], byte[]> currentElement = reader.getCurrent(); KV<byte[], byte[]> expectedElement = sourceOptions.genRecord(startPosition + 1).kv; assertEquals(expectedElement, currentElement); }
@Test public void lastElementShouldBeInclusive() throws IOException { int endPosition = 2; checkpoint = new SyntheticRecordsCheckpoint(0, endPosition); UnboundedSource.UnboundedReader<KV<byte[], byte[]>> reader = source.createReader(pipeline.getOptions(), checkpoint); reader.start(); reader.advance(); KV<byte[], byte[]> currentElement = reader.getCurrent(); KV<byte[], byte[]> expectedElement = sourceOptions.genRecord(endPosition).kv; assertEquals(expectedElement, currentElement); }