private void startAsynchronouslySplit() {
        if (!remainingTables.isEmpty()) {
            if (executor == null) {
                ThreadFactory threadFactory =
                        new ThreadFactoryBuilder().setNameFormat("snapshot-splitting").build();
                this.executor = Executors.newSingleThreadExecutor(threadFactory);
            }

            executor.submit(
                    () -> {
                        try {

                            for (TableId nextTable : remainingTables) {
                                // split the given table into chunks (snapshot splits)
                                Collection<MySqlSnapshotSplit> splits =
                                        chunkSplitter.generateSplits(nextTable);

                                if (!splits.isEmpty()) {
                                    tableSchemas.putAll(new HashMap<>(
                                            splits.iterator().next().getTableSchemas()));
                                }

                                final List<MySqlSchemalessSnapshotSplit> schemaLessSnapshotSplits =
                                        splits.stream()
                                                .map(MySqlSnapshotSplit::toSchemaLessSnapshotSplit)
                                                .collect(Collectors.toList());
                                synchronized (lock) {
                                    addNewlyAddedSplits(schemaLessSnapshotSplits);
                                    remainingTables.remove(nextTable);
                                    addAlreadyProcessedTablesIfNotExists(nextTable);
                                    lock.notify();
                                }
                            }
                        } catch (Exception e) {
                            LOG.error("asynchronously split exit with exception", e);
                        }
                    });
        }
    }
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