CountingSource { @Deprecated public static UnboundedSource<Long, CounterMark> unboundedWithTimestampFn( SerializableFunction<Long, Instant> timestampFn) { return new UnboundedCountingSource(0, 1, 1L, Duration.ZERO, timestampFn); } private  CountingSource(); @Deprecated static BoundedSource<Long> upTo(long numElements); @Deprecated static UnboundedSource<Long, CounterMark> unbounded(); @Deprecated static UnboundedSource<Long, CounterMark> unboundedWithTimestampFn(
      SerializableFunction<Long, Instant> timestampFn);  }
@Test @Category(NeedsRunner.class) public void testUnboundedSourceTimestamps() { long numElements = 1000; PCollection<Long> input = p.apply( Read.from(CountingSource.unboundedWithTimestampFn(new ValueAsTimestampFn())) .withMaxNumRecords(numElements)); addCountingAsserts(input, numElements); PCollection<Long> diffs = input .apply("TimestampDiff", ParDo.of(new ElementValueDiff())) .apply("DistinctTimestamps", Distinct.create()); PAssert.thatSingleton(diffs).isEqualTo(0L); p.run(); }
@Test public void testUnboundedSourceCheckpointMark() throws Exception { UnboundedSource<Long, CounterMark> source = CountingSource.unboundedWithTimestampFn(new ValueAsTimestampFn()); UnboundedReader<Long> reader = source.createReader(null, null); final long numToSkip = 3; assertTrue(reader.start()); for (long l = 0; l < numToSkip; ++l) { reader.advance(); } assertEquals(numToSkip, (long) reader.getCurrent()); assertEquals(numToSkip, reader.getCurrentTimestamp().getMillis()); CounterMark mark = CoderUtils.clone(source.getCheckpointMarkCoder(), (CounterMark) reader.getCheckpointMark()); reader = source.createReader(null, mark); assertTrue(reader.start()); assertEquals(numToSkip + 1, (long) reader.getCurrent()); assertEquals(numToSkip + 1, reader.getCurrentTimestamp().getMillis()); }