@Override
    public Map<String, String> tableOptions() {
        Map<String, String> options = super.tableOptions();
        options.put("connector", "mongodb-cdc-inlong");
        options.put("hosts", hosts);
        options.put("username", username);
        options.put("password", password);
        options.put("database", database);
        options.put("collection", collection);
        return options;
    }
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