public synchronized boolean autoEnterPureIncrementPhaseIfAllowed() {
        if (!enterPureIncrementPhase
                && maxSnapshotSplitsHighWatermark.compareTo(startupOffset) == 0) {
            split.asIncrementalSplit().getCompletedSnapshotSplitInfos().clear();
            enterPureIncrementPhase = true;
            return true;
        }
        return false;
    }
@Test
    public void testAutoEnterPureIncrementPhaseIfAllowed() {
        Offset startupOffset = new TestOffset(100);
        List<CompletedSnapshotSplitInfo> snapshotSplits = Collections.emptyList();
        IncrementalSplit split = createIncrementalSplit(startupOffset, snapshotSplits);
        IncrementalSplitState splitState = new IncrementalSplitState(split);
        Assertions.assertTrue(splitState.isEnterPureIncrementPhase());
        Assertions.assertFalse(splitState.autoEnterPureIncrementPhaseIfAllowed());

        startupOffset = new TestOffset(100);
        snapshotSplits =
                Stream.of(
                                createCompletedSnapshotSplitInfo(
                                        "test1", new TestOffset(100), new TestOffset(100)),
                                createCompletedSnapshotSplitInfo(
                                        "test2", new TestOffset(100), new TestOffset(100)))
                        .collect(Collectors.toList());
        split = createIncrementalSplit(startupOffset, snapshotSplits);
        splitState = new IncrementalSplitState(split);

        Assertions.assertFalse(splitState.isEnterPureIncrementPhase());
        Assertions.assertTrue(splitState.autoEnterPureIncrementPhaseIfAllowed());
        Assertions.assertTrue(splitState.isEnterPureIncrementPhase());
        Assertions.assertFalse(splitState.autoEnterPureIncrementPhaseIfAllowed());
        Assertions.assertTrue(splitState.isEnterPureIncrementPhase());

        startupOffset = new TestOffset(100);
        snapshotSplits =
                Stream.of(
                                createCompletedSnapshotSplitInfo(
                                        "test1", new TestOffset(100), new TestOffset(100)),
                                createCompletedSnapshotSplitInfo(
                                        "test2", new TestOffset(100), new TestOffset(101)))
                        .collect(Collectors.toList());
        split = createIncrementalSplit(startupOffset, snapshotSplits);
        splitState = new IncrementalSplitState(split);
        Assertions.assertFalse(splitState.isEnterPureIncrementPhase());
        Assertions.assertFalse(splitState.autoEnterPureIncrementPhaseIfAllowed());
    }