private MongoExtractNode buildComplexTypeMongoNode() {
        String[] fieldNames = new String[]{"name", "age"};
        FormatInfo[] fieldFormatInfos = new FormatInfo[]{new StringFormatInfo(), new StringFormatInfo()};
        List<FieldInfo> fields = Arrays.asList(
                new FieldInfo("info", new RowFormatInfo(fieldNames, fieldFormatInfos)),
                new MetaFieldInfo("proctime", MetaField.PROCESS_TIME),
                new MetaFieldInfo("database_name", MetaField.DATABASE_NAME),
                new MetaFieldInfo("collection_name", MetaField.COLLECTION_NAME),
                new MetaFieldInfo("op_ts", MetaField.OP_TS));
        return new MongoExtractNode("1", "mysql_input", fields,
                null, null, "test", "localhost:27017",
                "root", "inlong", "test");
    }
Test mongodb to kafka

@throws Exception The exception may throws when execute the case
inlong-sort/sort-core/src/test/java/org/apache/inlong/sort/parser/MongoExtractFlinkSqlParseTest.java