private MongoChangeStreamCursor<BsonDocument> openChangeStreamCursor(
            ChangeStreamDescriptor changeStreamDescriptor) {
        ChangeStreamOffset offset =
                new ChangeStreamOffset(streamSplit.getStartingOffset().getOffset());

        ChangeStreamIterable<Document> changeStreamIterable =
                getChangeStreamIterable(sourceConfig, changeStreamDescriptor);

        BsonDocument resumeToken = offset.getResumeToken();
        BsonTimestamp timestamp = offset.getTimestamp();

        if (resumeToken != null) {
            if (supportsStartAfter) {
                LOG.info("Open the change stream after the previous offset: {}", resumeToken);
                changeStreamIterable.startAfter(resumeToken);
            } else {
                LOG.info(
                        "Open the change stream after the previous offset using resumeAfter: {}",
                        resumeToken);
                changeStreamIterable.resumeAfter(resumeToken);
            }
        } else {
            if (supportsStartAtOperationTime) {
                LOG.info("Open the change stream at the timestamp: {}", timestamp);
                changeStreamIterable.startAtOperationTime(timestamp);
            } else {
                LOG.warn("Open the change stream of the latest offset");
            }
        }

        try {
            return (MongoChangeStreamCursor<BsonDocument>) changeStreamIterable.withDocumentClass(BsonDocument.class)
                    .cursor();
        } catch (MongoCommandException e) {
            if (e.getErrorCode() == FAILED_TO_PARSE_ERROR
                    || e.getErrorCode() == UNKNOWN_FIELD_ERROR) {
                if (e.getErrorMessage().contains("startAtOperationTime")) {
                    supportsStartAtOperationTime = false;
                    return openChangeStreamCursor(changeStreamDescriptor);
                } else if (e.getErrorMessage().contains("startAfter")) {
                    supportsStartAfter = false;
                    return openChangeStreamCursor(changeStreamDescriptor);
                } else {
                    LOG.error("Open change stream failed ", e);
                    throw new FlinkRuntimeException("Open change stream failed", e);
                }
            } else if (e.getErrorCode() == ILLEGAL_OPERATION_ERROR) {
                LOG.error(
                        "Illegal $changeStream operation: {} {}",
                        e.getErrorMessage(),
                        e.getErrorCode());
                throw new FlinkRuntimeException("Illegal $changeStream operation", e);
            } else if (e.getErrorCode() == UNAUTHORIZED_ERROR) {
                LOG.error(
                        "Unauthorized $changeStream operation: {} {}",
                        e.getErrorMessage(),
                        e.getErrorCode());
                throw new FlinkRuntimeException("Unauthorized $changeStream operation", e);
            } else {
                LOG.error("Open change stream failed ", e);
                throw new FlinkRuntimeException("Open change stream failed", e);
            }
        }
    }
The task to work for fetching data of MongoDB stream split .
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/reader/fetch/MongoDBStreamFetchTask.java