public static MySqlBinlogSplit filterOutdatedSplitInfos(
            MySqlBinlogSplit binlogSplit, Tables.TableFilter currentTableFilter) {
        Map<TableId, TableChange> filteredTableSchemas =
                binlogSplit.getTableSchemas().entrySet().stream()
                        .filter(entry -> currentTableFilter.isIncluded(entry.getKey()))
                        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

        Set<TableId> tablesToRemoveInFinishedSnapshotSplitInfos =
                binlogSplit.getFinishedSnapshotSplitInfos().stream()
                        .filter(i -> !currentTableFilter.isIncluded(i.getTableId()))
                        .map(split -> split.getTableId())
                        .collect(Collectors.toSet());
        if (tablesToRemoveInFinishedSnapshotSplitInfos.isEmpty()) {
            return new MySqlBinlogSplit(
                    binlogSplit.splitId,
                    binlogSplit.getStartingOffset(),
                    binlogSplit.getEndingOffset(),
                    binlogSplit.getFinishedSnapshotSplitInfos(),
                    filteredTableSchemas,
                    binlogSplit.totalFinishedSplitSize,
                    binlogSplit.isSuspended());
        }

        LOG.info(
                "Reader remove tables after restart: {}",
                tablesToRemoveInFinishedSnapshotSplitInfos);
        List<FinishedSnapshotSplitInfo> allFinishedSnapshotSplitInfos =
                binlogSplit.getFinishedSnapshotSplitInfos().stream()
                        .filter(
                                i ->
                                        !tablesToRemoveInFinishedSnapshotSplitInfos.contains(
                                                i.getTableId()))
                        .collect(Collectors.toList());
        return new MySqlBinlogSplit(
                binlogSplit.splitId,
                binlogSplit.getStartingOffset(),
                binlogSplit.getEndingOffset(),
                allFinishedSnapshotSplitInfos,
                filteredTableSchemas,
                binlogSplit.getTotalFinishedSplitSize()
                        - (binlogSplit.getFinishedSnapshotSplitInfos().size()
                                - allFinishedSnapshotSplitInfos.size()),
                binlogSplit.isSuspended());
    }
@Test
    public void filterOutdatedSplitInfos() {
        Map<TableId, TableChanges.TableChange> tableSchemas = new HashMap<>();

        // mock table1
        TableId tableId1 = new TableId("catalog1", null, "table1");

        TableChanges.TableChange tableChange1 =
                new TableChanges.TableChange(
                        TableChanges.TableChangeType.CREATE,
                        new MockTable(TableId.parse("catalog1.table1")));

        // mock table2
        TableId tableId2 = new TableId("catalog2", null, "table2");

        TableChanges.TableChange tableChange2 =
                new TableChanges.TableChange(
                        TableChanges.TableChangeType.CREATE,
                        new MockTable(TableId.parse("catalog2.table2")));
        tableSchemas.put(tableId1, tableChange1);
        tableSchemas.put(tableId2, tableChange2);
        MySqlBinlogSplit binlogSplit =
                new MySqlBinlogSplit(
                        "binlog-split",
                        BinlogOffset.ofLatest(),
                        null,
                        new ArrayList<>(),
                        tableSchemas,
                        0,
                        false);

        // case 1: only include table1
        Tables.TableFilter currentTableFilter = tableId -> tableId.table().equals("table1");

        MySqlBinlogSplit mySqlBinlogSplit =
                MySqlBinlogSplit.filterOutdatedSplitInfos(binlogSplit, currentTableFilter);
        Map<TableId, TableChanges.TableChange> filterTableSchemas =
                mySqlBinlogSplit.getTableSchemas();
        Assert.assertEquals(1, filterTableSchemas.size());
        Assert.assertEquals(tableChange1, filterTableSchemas.get(tableId1));

        // case 2: include all tables
        currentTableFilter = tableId -> tableId.table().startsWith("table");

        mySqlBinlogSplit =
                MySqlBinlogSplit.filterOutdatedSplitInfos(binlogSplit, currentTableFilter);
        filterTableSchemas = mySqlBinlogSplit.getTableSchemas();
        Assert.assertEquals(2, filterTableSchemas.size());
        Assert.assertEquals(tableChange1, filterTableSchemas.get(tableId1));
        Assert.assertEquals(tableChange2, filterTableSchemas.get(tableId2));
    }