@Override
  public ContinuousEnumerationResult planSplits(IcebergEnumeratorPosition lastPosition) {
    table.refresh();
    if (lastPosition != null) {
      return discoverIncrementalSplits(lastPosition);
    } else {
      return discoverInitialSplits();
    }
  }
@Test
  public void testIncrementalFromSnapshotTimestampWithInvalidIds() throws Exception {
    appendTwoSnapshots();

    long invalidSnapshotTimestampMs = snapshot2.timestampMillis() + 1000L;

    ScanContext scanContextWithInvalidSnapshotId =
        ScanContext.builder()
            .startingStrategy(StreamingStartingStrategy.INCREMENTAL_FROM_SNAPSHOT_TIMESTAMP)
            .startSnapshotTimestamp(invalidSnapshotTimestampMs)
            .build();

    ContinuousSplitPlannerImpl splitPlanner =
        new ContinuousSplitPlannerImpl(
            TABLE_RESOURCE.tableLoader().clone(), scanContextWithInvalidSnapshotId, null);

    assertThatThrownBy(() -> splitPlanner.planSplits(null))
        .isInstanceOf(IllegalArgumentException.class)
        .hasMessageStartingWith("Cannot find a snapshot after:");
  }