@Override
    public Collection<SnapshotSplit> split(SplitContext splitContext) {
        MongoClient mongoClient = splitContext.getMongoClient();
        TableId collectionId = splitContext.getCollectionId();
        int chunkSizeMB = splitContext.getChunkSizeMB();

        BsonDocument keyPattern = new BsonDocument(ID_FIELD, new BsonInt32(1));

        BsonDocument splitResult;
        try {
            splitResult = splitVector(mongoClient, collectionId, keyPattern, chunkSizeMB);
        } catch (MongoCommandException e) {
            if (e.getErrorCode() == UNAUTHORIZED_ERROR) {
                LOG.warn(
                        "Unauthorized to execute splitVector command: {}, fallback to SampleSplitter",
                        e.getErrorMessage());
            } else {
                LOG.warn(
                        "Execute splitVector command failed: {}, fallback to SampleSplitter",
                        e.getErrorMessage());
            }
            return SampleBucketSplitStrategy.INSTANCE.split(splitContext);
        }

        if (!isCommandSucceed(splitResult)) {
            LOG.warn(
                    "Could not calculate standalone splits: {}, fallback to SampleSplitter",
                    splitResult.getString("errmsg"));
            return SampleBucketSplitStrategy.INSTANCE.split(splitContext);
        }

        BsonArray splitKeys = splitResult.getArray("splitKeys");
        if (CollectionUtils.isEmpty(splitKeys)) {
            // documents size is less than chunk size, treat the entire collection as single chunk.
            return SingleSplitStrategy.INSTANCE.split(splitContext);
        }

        Map<TableId, TableChanges.TableChange> schema = new HashMap<>();
        schema.put(collectionId, collectionSchema(collectionId));

        RowType rowType = shardKeysToRowType(Collections.singleton(ID_FIELD));
        List<SnapshotSplit> snapshotSplits = new ArrayList<>(splitKeys.size() + 1);

        BsonValue lowerValue = BSON_MIN_KEY;
        for (int i = 0; i < splitKeys.size(); i++) {
            BsonValue splitKeyValue = splitKeys.get(i).asDocument().get(ID_FIELD);
            snapshotSplits.add(
                    new SnapshotSplit(
                            collectionId,
                            splitId(collectionId, i),
                            rowType,
                            boundOfId(lowerValue),
                            boundOfId(splitKeyValue),
                            null,
                            schema));
            lowerValue = splitKeyValue;
        }

        SnapshotSplit lastSplit =
                new SnapshotSplit(
                        collectionId,
                        splitId(collectionId, splitKeys.size()),
                        rowType,
                        boundOfId(lowerValue),
                        maxUpperBoundOfId(),
                        null,
                        schema);
        snapshotSplits.add(lastSplit);

        return snapshotSplits;
    }
The SplitVector Splitter.

<p>Uses the `SplitVector` command to generate chunks for a collection. eg. <code>
db.runCommand({splitVector:"inventory.products", keyPattern:{_id:1}, maxChunkSize:64})</code>
Requires `splitVector` privilege.
Copy from com.ververica:flink-connector-mongodb-cdc:2.3.0.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/assigners/splitters/SplitVectorSplitStrategy.java