@Override
  public IcebergEnumeratorState snapshotState(long checkpointId) {
    return new IcebergEnumeratorState(
        enumeratorPosition.get(), assigner.state(), enumerationHistory.snapshot());
  }
@Test
  public void testTransientPlanningErrorsWithSuccessfulRetry() throws Exception {
    TestingSplitEnumeratorContext<IcebergSourceSplit> enumeratorContext =
        new TestingSplitEnumeratorContext<>(4);
    ScanContext scanContext =
        ScanContext.builder()
            .streaming(true)
            .startingStrategy(StreamingStartingStrategy.INCREMENTAL_FROM_EARLIEST_SNAPSHOT)
            .maxPlanningSnapshotCount(1)
            .maxAllowedPlanningFailures(2)
            .build();
    ManualContinuousSplitPlanner splitPlanner = new ManualContinuousSplitPlanner(scanContext, 1);
    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);

    // Trigger a planning and check that no splits returned due to the planning error
    enumeratorContext.triggerAllActions();
    assertThat(enumerator.snapshotState(2).pendingSplits()).isEmpty();

    // Second scan planning should succeed and discover the expected splits
    enumeratorContext.triggerAllActions();
    Collection<IcebergSourceSplitState> pendingSplits = enumerator.snapshotState(3).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);
  }