@Override
  public IcebergEnumeratorState snapshotState(long checkpointId) {
    return new IcebergEnumeratorState(
        enumeratorPosition.get(), assigner.state(), enumerationHistory.snapshot());
  }
@Test
  public void testThrottlingDiscovery() throws Exception {
    // create 10 splits
    List<IcebergSourceSplit> splits =
        SplitHelpers.createSplitsFromTransientHadoopTable(temporaryFolder, 10, 1);

    TestingSplitEnumeratorContext<IcebergSourceSplit> enumeratorContext =
        new TestingSplitEnumeratorContext<>(4);
    ScanContext scanContext =
        ScanContext.builder()
            .streaming(true)
            .startingStrategy(StreamingStartingStrategy.INCREMENTAL_FROM_EARLIEST_SNAPSHOT)
            // discover one snapshot at a time
            .maxPlanningSnapshotCount(1)
            .build();
    ManualContinuousSplitPlanner splitPlanner = new ManualContinuousSplitPlanner(scanContext, 0);
    ContinuousIcebergEnumerator enumerator =
        createEnumerator(enumeratorContext, scanContext, splitPlanner);

    // register reader-2, and let it request a split
    enumeratorContext.registerReader(2, "localhost");
    enumerator.addReader(2);
    enumerator.handleSourceEvent(2, new SplitRequestEvent());

    // add splits[0] to the planner for next discovery
    splitPlanner.addSplits(Arrays.asList(splits.get(0)));
    enumeratorContext.triggerAllActions();

    // because discovered split was assigned to reader, pending splits should be empty
    assertThat(enumerator.snapshotState(1).pendingSplits()).isEmpty();
    // split assignment to reader-2 should contain splits[0, 1)
    assertThat(enumeratorContext.getSplitAssignments().get(2).getAssignedSplits())
        .containsExactlyElementsOf(splits.subList(0, 1));

    // add the remaining 9 splits (one for every snapshot)
    // run discovery cycles while reader-2 still processing the splits[0]
    for (int i = 1; i < 10; ++i) {
      splitPlanner.addSplits(Arrays.asList(splits.get(i)));
      enumeratorContext.triggerAllActions();
    }

    // can only discover up to 3 snapshots/splits
    assertThat(enumerator.snapshotState(2).pendingSplits()).hasSize(3);
    // split assignment to reader-2 should be splits[0, 1)
    assertThat(enumeratorContext.getSplitAssignments().get(2).getAssignedSplits())
        .containsExactlyElementsOf(splits.subList(0, 1));

    // now reader-2 finished splits[0]
    enumerator.handleSourceEvent(2, new SplitRequestEvent(Arrays.asList(splits.get(0).splitId())));
    enumeratorContext.triggerAllActions();
    // still have 3 pending splits. After assigned splits[1] to reader-2, one more split was
    // discovered and added.
    assertThat(enumerator.snapshotState(3).pendingSplits()).hasSize(3);
    // split assignment to reader-2 should be splits[0, 2)
    assertThat(enumeratorContext.getSplitAssignments().get(2).getAssignedSplits())
        .containsExactlyElementsOf(splits.subList(0, 2));

    // run 3 more split discovery cycles
    for (int i = 0; i < 3; ++i) {
      enumeratorContext.triggerAllActions();
    }

    // no more splits are discovered due to throttling
    assertThat(enumerator.snapshotState(4).pendingSplits()).hasSize(3);
    // split assignment to reader-2 should still be splits[0, 2)
    assertThat(enumeratorContext.getSplitAssignments().get(2).getAssignedSplits())
        .containsExactlyElementsOf(splits.subList(0, 2));

    // now reader-2 finished splits[1]
    enumerator.handleSourceEvent(2, new SplitRequestEvent(Arrays.asList(splits.get(1).splitId())));
    enumeratorContext.triggerAllActions();
    // still have 3 pending splits. After assigned new splits[2] to reader-2, one more split was
    // discovered and added.
    assertThat(enumerator.snapshotState(5).pendingSplits()).hasSize(3);
    // split assignment to reader-2 should be splits[0, 3)
    assertThat(enumeratorContext.getSplitAssignments().get(2).getAssignedSplits())
        .containsExactlyElementsOf(splits.subList(0, 3));
  }