@Override
    public void execute(Context context) throws Exception {
        MongoDBFetchTaskContext taskContext = (MongoDBFetchTaskContext) context;
        this.sourceConfig = taskContext.getSourceConfig();

        ChangeStreamDescriptor descriptor = taskContext.getChangeStreamDescriptor();
        ChangeEventQueue<DataChangeEvent> queue = taskContext.getQueue();

        MongoClient mongoClient = clientFor(sourceConfig);
        MongoChangeStreamCursor<BsonDocument> changeStreamCursor =
                openChangeStreamCursor(descriptor);
        HeartbeatManager heartbeatManager = openHeartbeatManagerIfNeeded(changeStreamCursor);

        final long startPoll = time.milliseconds();
        long nextUpdate = startPoll + sourceConfig.getPollAwaitTimeMillis();
        this.taskRunning = true;
        try {
            while (taskRunning) {
                Optional<BsonDocument> next = Optional.ofNullable(changeStreamCursor.tryNext());
                SourceRecord changeRecord = null;
                if (!next.isPresent()) {
                    long untilNext = nextUpdate - time.milliseconds();
                    if (untilNext > 0) {
                        LOG.debug("Waiting {} ms to poll change records", untilNext);
                        time.sleep(untilNext);
                        continue;
                    }

                    if (heartbeatManager != null) {
                        changeRecord =
                                heartbeatManager
                                        .heartbeat()
                                        .map(this::normalizeHeartbeatRecord)
                                        .orElse(null);
                    }
                } else {
                    BsonDocument changeStreamDocument = next.get();
                    MongoNamespace namespace = getMongoNamespace(changeStreamDocument);

                    BsonDocument resumeToken = changeStreamDocument.getDocument(ID_FIELD);
                    BsonDocument valueDocument =
                            normalizeChangeStreamDocument(changeStreamDocument);

                    LOG.trace("Adding {} to {}", valueDocument, namespace.getFullName());

                    changeRecord =
                            createSourceRecord(
                                    createPartitionMap(
                                            sourceConfig.getHosts(),
                                            namespace.getDatabaseName(),
                                            namespace.getCollectionName()),
                                    createSourceOffsetMap(resumeToken, false),
                                    namespace.getFullName(),
                                    changeStreamDocument.getDocument(ID_FIELD),
                                    valueDocument);
                }

                if (changeRecord != null) {
                    queue.enqueue(new DataChangeEvent(changeRecord));
                }

                if (isBoundedRead()) {
                    ChangeStreamOffset currentOffset;
                    if (changeRecord != null) {
                        currentOffset = new ChangeStreamOffset(getResumeToken(changeRecord));
                    } else {
                        // Heartbeat is not turned on or there is no update event
                        currentOffset = new ChangeStreamOffset(getCurrentClusterTime(mongoClient));
                    }

                    // Reach the high watermark, the binlog fetcher should be finished
                    if (currentOffset.isAtOrAfter(streamSplit.getEndingOffset())) {
                        // send watermark end event
                        SourceRecord watermark =
                                WatermarkEvent.create(
                                        createWatermarkPartitionMap(descriptor.toString()),
                                        WATERMARK_TOPIC_NAME,
                                        streamSplit.splitId(),
                                        WatermarkKind.END,
                                        currentOffset);

                        queue.enqueue(new DataChangeEvent(watermark));
                        break;
                    }
                }
            }
        } catch (Exception e) {
            LOG.error("Poll change stream records failed ", e);
            throw e;
        } finally {
            taskRunning = false;
            if (changeStreamCursor != null) {
                changeStreamCursor.close();
            }
        }
    }
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