@Override
    protected void onSplitFinished(Map<String, MySqlSplitState> finishedSplitIds) {
        boolean requestNextSplit = true;
        for (MySqlSplitState mySqlSplitState : finishedSplitIds.values()) {
            MySqlSplit mySqlSplit = mySqlSplitState.toMySqlSplit();
            if (mySqlSplit.isBinlogSplit()) {
                LOG.info(
                        "binlog split reader suspended due to newly added table, offset {}",
                        mySqlSplitState.asBinlogSplitState().getStartingOffset());

                mySqlSourceReaderContext.resetStopBinlogSplitReader();
                suspendedBinlogSplit = toSuspendedBinlogSplit(mySqlSplit.asBinlogSplit());
                context.sendSourceEventToCoordinator(new SuspendBinlogReaderAckEvent());
                // do not request next split when the reader is suspended, the suspended reader will
                // automatically request the next split after it has been wakeup
                requestNextSplit = false;
            } else {
                finishedUnackedSplits.put(mySqlSplit.splitId(), mySqlSplit.asSnapshotSplit());
            }
        }
        reportFinishedSnapshotSplitsIfNeed();
        if (requestNextSplit) {
            context.sendSplitRequest();
        }
    }
The source reader for MySQL source splits.
copy from {@link com.ververica.cdc.connectors.mysql.source.reader.MySqlSourceReader}
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/mysql/source/reader/MySqlSourceReader.java