@Override
    public Set<MetaField> supportedMetaFields() {
        return EnumSet.of(MetaField.PROCESS_TIME, MetaField.COLLECTION_NAME,
                MetaField.DATABASE_NAME, MetaField.OP_TS,
                MetaField.DATA_DEBEZIUM, MetaField.DATA_BYTES_DEBEZIUM,
                MetaField.DATA_CANAL, MetaField.DATA_BYTES_CANAL);
    }
the primary key must be "_id"

@see <a href="https://ververica.github.io/flink-cdc-connectors/release-2.1/content/connectors/mongodb-cdc.html#">MongoDB CDC Connector</a>
inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/extract/MongoExtractNode.java