@Override
    public void notifyCheckpointComplete(long completedCheckpointId) {
        synchronized (uploadedSstFiles) {
            // FLINK-23949: materializedSstFiles.keySet().contains(completedCheckpointId) make sure
            // the notified checkpointId is not a savepoint, otherwise next checkpoint will
            // degenerate into a full checkpoint
            if (completedCheckpointId > lastCompletedCheckpointId
                    && uploadedSstFiles.containsKey(completedCheckpointId)) {
                uploadedSstFiles
                        .keySet()
                        .removeIf(checkpointId -> checkpointId < completedCheckpointId);
                lastCompletedCheckpointId = completedCheckpointId;
            }
        }
    }
@Test
    void testCheckpointIsIncremental() throws Exception {

        try (CloseableRegistry closeableRegistry = new CloseableRegistry();
                RocksIncrementalSnapshotStrategy<?> checkpointSnapshotStrategy =
                        createSnapshotStrategy()) {
            FsCheckpointStreamFactory checkpointStreamFactory = createFsCheckpointStreamFactory();

            // make and notify checkpoint with id 1
            snapshot(1L, checkpointSnapshotStrategy, checkpointStreamFactory, closeableRegistry);
            checkpointSnapshotStrategy.notifyCheckpointComplete(1L);

            // notify savepoint with id 2
            checkpointSnapshotStrategy.notifyCheckpointComplete(2L);

            // make checkpoint with id 3
            IncrementalRemoteKeyedStateHandle incrementalRemoteKeyedStateHandle3 =
                    snapshot(
                            3L,
                            checkpointSnapshotStrategy,
                            checkpointStreamFactory,
                            closeableRegistry);

            // If 3rd checkpoint's full size > checkpointed size, it means 3rd checkpoint is
            // incremental.
            assertThat(incrementalRemoteKeyedStateHandle3.getStateSize())
                    .isGreaterThan(incrementalRemoteKeyedStateHandle3.getCheckpointedSize());
        }
    }