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

    // find an invalid snapshotId
    long invalidSnapshotId = 0L;
    while (invalidSnapshotId == snapshot1.snapshotId()
        || invalidSnapshotId == snapshot2.snapshotId()) {
      invalidSnapshotId++;
    }

    ScanContext scanContextWithInvalidSnapshotId =
        ScanContext.builder()
            .startingStrategy(StreamingStartingStrategy.INCREMENTAL_FROM_SNAPSHOT_ID)
            .startSnapshotId(invalidSnapshotId)
            .build();

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

    assertThatThrownBy(() -> splitPlanner.planSplits(null))
        .isInstanceOf(IllegalArgumentException.class)
        .hasMessage("Start snapshot id not found in history: " + invalidSnapshotId);
  }