@Override
    public Optional<MySqlSplit> getNext() {
        synchronized (lock) {
            if (!remainingSplits.isEmpty()) {
                // return remaining splits firstly
                Iterator<MySqlSchemalessSnapshotSplit> iterator = remainingSplits.iterator();
                MySqlSchemalessSnapshotSplit split = iterator.next();
                remainingSplits.remove(split);
                assignedSplits.put(split.splitId(), split);
                return Optional.of(
                        split.toMySqlSnapshotSplit(tableSchemas.get(split.getTableId())));
            } else if (!remainingTables.isEmpty()) {
                try {
                    // wait for the asynchronous split to complete
                    lock.wait();
                } catch (InterruptedException e) {
                    throw new FlinkRuntimeException(
                            "InterruptedException while waiting for asynchronously snapshot split");
                }
                return getNext();
            } else {
                closeExecutorService();
                return Optional.empty();
            }
        }
    }
A {@link MySqlSplitAssigner} that splits tables into small chunk splits based on primary key
range and chunk size.

@see MySqlSourceOptions#SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/MySqlSnapshotSplitAssigner.java