@Override
    public List<SourceRecord> formatMessageTimestamp(Collection<SourceRecord> snapshotRecords) {
        return snapshotRecords.stream()
                .peek(
                        record -> {
                            Struct value = (Struct) record.value();
                            // set message timestamp (source.ts_ms) to 0L
                            Struct source =
                                    new Struct(
                                            value.schema()
                                                    .field(MongoDBEnvelope.SOURCE_FIELD)
                                                    .schema());
                            source.put(MongoDBEnvelope.TIMESTAMP_KEY_FIELD, 0L);
                            source.put(MongoDBEnvelope.SNAPSHOT_KEY_FIELD, "true");
                            value.put(MongoDBEnvelope.SOURCE_FIELD, source);
                        })
                .collect(Collectors.toList());
    }
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