public static Builder builder() {
    return new Builder();
  }
@Test
  void testFilterInstantsWithRange() throws IOException {
    Configuration conf = TestConfigurations.getDefaultConf(basePath);
    conf.set(FlinkOptions.READ_STREAMING_SKIP_CLUSTERING, true);
    conf.set(FlinkOptions.TABLE_TYPE, FlinkOptions.TABLE_TYPE_MERGE_ON_READ);
    metaClient = HoodieTestUtils.init(basePath, HoodieTableType.MERGE_ON_READ);

    HoodieActiveTimeline timeline = metaClient.getActiveTimeline();
    HoodieInstant commit1 = new HoodieInstant(HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION, "1");
    HoodieInstant commit2 = new HoodieInstant(HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION, "2");
    HoodieInstant commit3 = new HoodieInstant(HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION, "3");
    timeline.createCompleteInstant(commit1);
    timeline.createCompleteInstant(commit2);
    timeline.createCompleteInstant(commit3);
    timeline = metaClient.reloadActiveTimeline();

    Map<String, String> completionTimeMap = timeline.filterCompletedInstants().getInstantsAsStream()
        .collect(Collectors.toMap(HoodieInstant::getTimestamp, HoodieInstant::getCompletionTime));

    IncrementalQueryAnalyzer analyzer1 = IncrementalQueryAnalyzer.builder()
        .metaClient(metaClient)
        .rangeType(InstantRange.RangeType.OPEN_CLOSED)
        .startTime(completionTimeMap.get("1"))
        .skipClustering(true)
        .build();
    // previous read iteration read till instant time "1", next read iteration should return ["2", "3"]
    List<HoodieInstant> activeInstants1 = analyzer1.analyze().getActiveInstants();
    assertEquals(2, activeInstants1.size());
    assertIterableEquals(Arrays.asList(commit2, commit3), activeInstants1);

    // simulate first iteration cycle with read from the LATEST commit
    IncrementalQueryAnalyzer analyzer2 = IncrementalQueryAnalyzer.builder()
        .metaClient(metaClient)
        .rangeType(InstantRange.RangeType.CLOSED_CLOSED)
        .skipClustering(true)
        .build();
    List<HoodieInstant> activeInstants2 = analyzer2.analyze().getActiveInstants();
    assertEquals(1, activeInstants2.size());
    assertIterableEquals(Collections.singletonList(commit3), activeInstants2);

    // specifying a start and end commit
    IncrementalQueryAnalyzer analyzer3 = IncrementalQueryAnalyzer.builder()
        .metaClient(metaClient)
        .rangeType(InstantRange.RangeType.CLOSED_CLOSED)
        .startTime(completionTimeMap.get("1"))
        .endTime(completionTimeMap.get("3"))
        .skipClustering(true)
        .build();
    List<HoodieInstant> activeInstants3 = analyzer3.analyze().getActiveInstants();
    assertEquals(3, activeInstants3.size());
    assertIterableEquals(Arrays.asList(commit1, commit2, commit3), activeInstants3);

    // add an inflight instant which should be excluded
    HoodieInstant commit4 = new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.COMPACTION_ACTION, "4");
    timeline.createNewInstant(commit4);
    timeline = metaClient.reloadActiveTimeline();
    assertEquals(4, timeline.getInstants().size());
    List<HoodieInstant> activeInstants4 = analyzer3.analyze().getActiveInstants();
    assertEquals(3, activeInstants4.size());
  }