@NotNull
    List<PartitionIdentifier> choosePartitionsToCompact(@NotNull Set<PartitionIdentifier> excludes,
            @NotNull Set<Long> excludeTables) {
        return choosePartitionsToCompact(excludeTables).stream().filter(p -> !excludes.contains(p)).collect(Collectors.toList());
    }
@Test
    public void testChoosePartitionsToCompact() {
        Config.lake_compaction_selector = "SimpleSelector";
        Config.lake_compaction_sorter = "RandomSorter";
        CompactionMgr compactionManager = new CompactionMgr();

        PartitionIdentifier partition1 = new PartitionIdentifier(1, 2, 3);
        PartitionIdentifier partition2 = new PartitionIdentifier(1, 2, 4);

        Set<Long> excludeTables = new HashSet<>();
        for (int i = 1; i <= Config.lake_compaction_simple_selector_threshold_versions - 1; i++) {
            compactionManager.handleLoadingFinished(partition1, i, System.currentTimeMillis(), null);
            compactionManager.handleLoadingFinished(partition2, i, System.currentTimeMillis(), null);
            Assert.assertEquals(0, compactionManager.choosePartitionsToCompact(excludeTables).size());
        }
        compactionManager.handleLoadingFinished(partition1, Config.lake_compaction_simple_selector_threshold_versions,
                System.currentTimeMillis(), null);
        List<PartitionIdentifier> compactionList = compactionManager.choosePartitionsToCompact(excludeTables);
        Assert.assertEquals(1, compactionList.size());
        Assert.assertSame(partition1, compactionList.get(0));

        Assert.assertEquals(compactionList, compactionManager.choosePartitionsToCompact(excludeTables));

        compactionManager.handleLoadingFinished(partition2, Config.lake_compaction_simple_selector_threshold_versions,
                System.currentTimeMillis(), null);

        compactionList = compactionManager.choosePartitionsToCompact(excludeTables);
        Assert.assertEquals(2, compactionList.size());
        Assert.assertTrue(compactionList.contains(partition1));
        Assert.assertTrue(compactionList.contains(partition2));

        compactionList = compactionManager.choosePartitionsToCompact(Collections.singleton(partition1), excludeTables);
        Assert.assertEquals(1, compactionList.size());
        Assert.assertSame(partition2, compactionList.get(0));

        compactionManager.enableCompactionAfter(partition1, 5000);
        compactionManager.enableCompactionAfter(partition2, 5000);
        compactionList = compactionManager.choosePartitionsToCompact(excludeTables);
        Assert.assertEquals(0, compactionList.size());

        compactionManager.enableCompactionAfter(partition1, 0);
        compactionManager.enableCompactionAfter(partition2, 0);
        compactionManager.removePartition(partition1);
        compactionList = compactionManager.choosePartitionsToCompact(excludeTables);
        Assert.assertEquals(1, compactionList.size());
        Assert.assertSame(partition2, compactionList.get(0));
    }