private BsonDocument normalizeChangeStreamDocument(BsonDocument changeStreamDocument) {
        // _id: primary key of change document.
        changeStreamDocument.put(ID_FIELD, normalizeKeyDocument(changeStreamDocument));

        // ts_ms: It indicates the time at which the reader processed the event.
        changeStreamDocument.put(TIMESTAMP_KEY_FIELD, new BsonInt64(System.currentTimeMillis()));

        // source
        BsonDocument source = new BsonDocument();
        source.put(SNAPSHOT_KEY_FIELD, new BsonString("false"));

        // cluster time after MongoDB version 4.0.
        if (!changeStreamDocument.containsKey(CLUSTER_TIME_FIELD)) {
            // Legacy ts_ms of MongoDB version [3.6, 4.0)
            if (changeStreamDocument.containsKey(TIMESTAMP_KEY_FIELD)) {
                long timestampValue = changeStreamDocument.getInt64(TIMESTAMP_KEY_FIELD).getValue();
                BsonTimestamp legacyTimestamp = bsonTimestampFromEpochMillis(timestampValue);
                changeStreamDocument.put(CLUSTER_TIME_FIELD, legacyTimestamp);
            } else { // Fallback to current timestamp.
                LOG.warn(
                        "Cannot extract clusterTime from change stream event, fallback to current timestamp.");
                changeStreamDocument.put(CLUSTER_TIME_FIELD, currentBsonTimestamp());
            }
        }

        // source.ts_ms
        // It indicates the time that the change was made in the database. If the record is read
        // from snapshot of the table instead of the change stream, the value is always 0.
        BsonTimestamp clusterTime = changeStreamDocument.getTimestamp(CLUSTER_TIME_FIELD);
        Instant clusterInstant = Instant.ofEpochSecond(clusterTime.getTime());
        source.put(TIMESTAMP_KEY_FIELD, new BsonInt64(clusterInstant.toEpochMilli()));
        changeStreamDocument.put(SOURCE_FIELD, source);

        return changeStreamDocument;
    }
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