@Override
    public boolean isRecordBetween(SourceRecord record, Object[] splitStart, Object[] splitEnd) {
        BsonDocument documentKey = getDocumentKey(record);
        // In the case of a compound index, we can also agree to only compare the range of the first
        // key.
        BsonDocument splitKeys = (BsonDocument) splitStart[0];
        String firstKey = splitKeys.getFirstKey();
        BsonValue keyValue = documentKey.get(firstKey);
        BsonValue lowerBound = ((BsonDocument) splitStart[1]).get(firstKey);
        BsonValue upperBound = ((BsonDocument) splitEnd[1]).get(firstKey);

        // for all range
        if (lowerBound.getBsonType() == BsonType.MIN_KEY
                && upperBound.getBsonType() == BsonType.MAX_KEY) {
            return true;
        }

        // keyValue -> [lower, upper)
        if (compareBsonValue(lowerBound, keyValue) <= 0
                && compareBsonValue(keyValue, upperBound) < 0) {
            return true;
        }

        return false;
    }
The context for fetch task that fetching data of snapshot split from MongoDB data source.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/reader/fetch/MongoDBFetchTaskContext.java