@Override
    public Optional<SourceSplitBase> getNext() {
        if (!remainingSplits.isEmpty()) {
            // return remaining splits firstly
            Iterator<SchemalessSnapshotSplit> iterator = remainingSplits.iterator();
            SchemalessSnapshotSplit split = iterator.next();
            iterator.remove();
            assignedSplits.put(split.splitId(), split);
            return Optional.of(split.toSnapshotSplit(tableSchemas.get(split.getTableId())));
        } else {
            // it's turn for new table
            TableId nextTable = remainingTables.pollFirst();
            if (nextTable != null) {
                // split the given table into chunks (snapshot splits)
                Collection<SnapshotSplit> splits = chunkSplitter.generateSplits(nextTable);
                final Map<TableId, TableChanges.TableChange> tableSchema = new HashMap<>();
                if (!splits.isEmpty()) {
                    tableSchema.putAll(splits.iterator().next().getTableSchemas());
                }
                final List<SchemalessSnapshotSplit> schemalessSnapshotSplits =
                        splits.stream()
                                .map(SnapshotSplit::toSchemalessSnapshotSplit)
                                .collect(Collectors.toList());
                remainingSplits.addAll(schemalessSnapshotSplits);
                tableSchemas.putAll(tableSchema);
                alreadyProcessedTables.add(nextTable);
                return getNext();
            } else {
                return Optional.empty();
            }
        }
    }
Assigner for snapshot split.
Copy from com.ververica:flink-cdc-base:2.3.0.
inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/SnapshotSplitAssigner.java