@Deprecated
  public static UnboundedSource<Long, CounterMark> unboundedWithTimestampFn(
      SerializableFunction<Long, Instant> timestampFn) {
    return new UnboundedCountingSource(0, 1, 1L, Duration.ZERO, timestampFn);
  }
@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());

    // Advance the source numToSkip elements and manually save state.
    for (long l = 0; l < numToSkip; ++l) {
      reader.advance();
    }

    // Confirm that we get the expected element in sequence before checkpointing.
    assertEquals(numToSkip, (long) reader.getCurrent());
    assertEquals(numToSkip, reader.getCurrentTimestamp().getMillis());

    // Checkpoint and restart, and confirm that the source continues correctly.
    CounterMark mark =
        CoderUtils.clone(source.getCheckpointMarkCoder(), (CounterMark) reader.getCheckpointMark());
    reader = source.createReader(null, mark);
    assertTrue(reader.start());

    // Confirm that we get the next element in sequence.
    assertEquals(numToSkip + 1, (long) reader.getCurrent());
    assertEquals(numToSkip + 1, reader.getCurrentTimestamp().getMillis());
  }