@Override
  public IcebergEnumeratorState snapshotState(long checkpointId) {
    return new IcebergEnumeratorState(
        enumeratorPosition.get(), assigner.state(), enumerationHistory.snapshot());
  }
@Test
  public void testPlanningIgnoringErrors() throws Exception {
    int expectedFailures = 3;
    TestingSplitEnumeratorContext<IcebergSourceSplit> enumeratorContext =
        new TestingSplitEnumeratorContext<>(4);
    ScanContext scanContext =
        ScanContext.builder()
            .streaming(true)
            .startingStrategy(StreamingStartingStrategy.INCREMENTAL_FROM_EARLIEST_SNAPSHOT)
            .maxPlanningSnapshotCount(1)
            .maxAllowedPlanningFailures(-1)
            .build();
    ManualContinuousSplitPlanner splitPlanner =
        new ManualContinuousSplitPlanner(scanContext, expectedFailures);
    ContinuousIcebergEnumerator enumerator =
        createEnumerator(enumeratorContext, scanContext, splitPlanner);

    // Make one split available and trigger the periodic discovery
    List<IcebergSourceSplit> splits =
        SplitHelpers.createSplitsFromTransientHadoopTable(temporaryFolder, 1, 1);
    splitPlanner.addSplits(splits);

    Collection<IcebergSourceSplitState> pendingSplits;
    // Can not discover the new split with planning failures
    for (int i = 0; i < expectedFailures; ++i) {
      enumeratorContext.triggerAllActions();
      pendingSplits = enumerator.snapshotState(i).pendingSplits();
      assertThat(pendingSplits).isEmpty();
    }

    // Discovered the new split after a successful scan planning
    enumeratorContext.triggerAllActions();
    pendingSplits = enumerator.snapshotState(expectedFailures + 1).pendingSplits();
    assertThat(pendingSplits).hasSize(1);
    IcebergSourceSplitState pendingSplit = pendingSplits.iterator().next();
    assertThat(pendingSplit.split().splitId()).isEqualTo(splits.get(0).splitId());
    assertThat(pendingSplit.status()).isEqualTo(IcebergSourceSplitStatus.UNASSIGNED);
  }