@Override
    public void execute(Context context) throws Exception {
        MongoDBFetchTaskContext taskContext = (MongoDBFetchTaskContext) context;
        MongoDBSourceConfig sourceConfig = taskContext.getSourceConfig();
        MongoDBDialect dialect = taskContext.getDialect();
        ChangeEventQueue<DataChangeEvent> changeEventQueue = taskContext.getQueue();

        taskRunning = true;
        TableId collectionId = snapshotSplit.getTableId();

        final ChangeStreamOffset lowWatermark = dialect.displayCurrentOffset(sourceConfig);
        LOG.info(
                "Snapshot step 1 - Determining low watermark {} for split {}",
                lowWatermark,
                snapshotSplit);

        changeEventQueue.enqueue(
                new DataChangeEvent(
                        WatermarkEvent.create(
                                createWatermarkPartitionMap(collectionId.identifier()),
                                WATERMARK_TOPIC_NAME,
                                snapshotSplit.splitId(),
                                WatermarkKind.LOW,
                                lowWatermark)));

        LOG.info("Snapshot step 2 - Snapshotting data");
        MongoCursor<RawBsonDocument> cursor = null;
        try {
            MongoClient mongoClient = clientFor(sourceConfig);
            MongoCollection<RawBsonDocument> collection =
                    collectionFor(mongoClient, collectionId, RawBsonDocument.class);

            // Using min and max operation to perform a specific index scan
            // See: https://www.mongodb.com/docs/manual/reference/method/cursor.min/
            cursor =
                    collection
                            .find()
                            .min((BsonDocument) snapshotSplit.getSplitStart()[1])
                            .max((BsonDocument) snapshotSplit.getSplitEnd()[1])
                            .hint((BsonDocument) snapshotSplit.getSplitStart()[0])
                            .batchSize(sourceConfig.getBatchSize())
                            .noCursorTimeout(true)
                            .cursor();

            BsonDocument keyDocument, valueDocument;
            while (cursor.hasNext()) {
                if (!taskRunning) {
                    throw new InterruptedException(
                            "Interrupted while snapshotting collection "
                                    + collectionId.identifier());
                }

                valueDocument = normalizeSnapshotDocument(collectionId, cursor.next());
                keyDocument = new BsonDocument(ID_FIELD, valueDocument.get(ID_FIELD));

                SourceRecord snapshotRecord =
                        createSourceRecord(
                                createPartitionMap(
                                        sourceConfig.getHosts(),
                                        collectionId.catalog(),
                                        collectionId.table()),
                                createSourceOffsetMap(keyDocument.getDocument(ID_FIELD), true),
                                collectionId.identifier(),
                                keyDocument,
                                valueDocument);

                changeEventQueue.enqueue(new DataChangeEvent(snapshotRecord));
            }

            ChangeStreamOffset highWatermark = dialect.displayCurrentOffset(sourceConfig);

            LOG.info(
                    "Snapshot step 3 - Determining high watermark {} for split {}",
                    highWatermark,
                    snapshotSplit);
            changeEventQueue.enqueue(
                    new DataChangeEvent(
                            WatermarkEvent.create(
                                    createWatermarkPartitionMap(collectionId.identifier()),
                                    WATERMARK_TOPIC_NAME,
                                    snapshotSplit.splitId(),
                                    WatermarkKind.HIGH,
                                    highWatermark)));

            LOG.info(
                    "Snapshot step 4 - Back fill stream split for snapshot split {}",
                    snapshotSplit);
            final StreamSplit backfillStreamSplit =
                    createBackfillStreamSplit(lowWatermark, highWatermark);

            // optimization that skip the stream read when the low watermark equals high watermark
            final boolean streamBackfillRequired =
                    backfillStreamSplit
                            .getEndingOffset()
                            .isAfter(backfillStreamSplit.getStartingOffset());

            if (!streamBackfillRequired) {
                changeEventQueue.enqueue(
                        new DataChangeEvent(
                                WatermarkEvent.create(
                                        createWatermarkPartitionMap(collectionId.identifier()),
                                        WATERMARK_TOPIC_NAME,
                                        backfillStreamSplit.splitId(),
                                        WatermarkKind.END,
                                        backfillStreamSplit.getEndingOffset())));
            } else {
                MongoDBStreamFetchTask backfillStreamTask =
                        new MongoDBStreamFetchTask(backfillStreamSplit);
                backfillStreamTask.execute(taskContext);
            }

            taskRunning = false;
        } catch (Exception e) {
            taskRunning = false;
            LOG.error(
                    String.format(
                            "Execute snapshot read subtask for mongo split %s fail", snapshotSplit),
                    e);
            throw e;
        } finally {
            if (cursor != null) {
                cursor.close();
            }
        }
    }
The task to work for fetching data of MongoDB collection snapshot 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/MongoDBScanFetchTask.java