private KafkaLoadNode buildComplexTypeKafkaLoadNode() {
        List<FieldInfo> fields = Arrays.asList(
                new FieldInfo("name", new StringFormatInfo()),
                new FieldInfo("_id", new StringFormatInfo()),
                new FieldInfo("proctime", new TimestampFormatInfo()),
                new FieldInfo("database_name", new StringFormatInfo()),
                new FieldInfo("collection_name", new StringFormatInfo()),
                new FieldInfo("op_ts", new TimestampFormatInfo()));
        List<FieldRelation> relations = Arrays.asList(
                new FieldRelation(new FieldInfo("info.name", new StringFormatInfo()),
                        new FieldInfo("name", new StringFormatInfo())),
                new FieldRelation(new FieldInfo("_id", new StringFormatInfo()),
                        new FieldInfo("_id", new StringFormatInfo())),
                new FieldRelation(new FieldInfo("proctime", new TimestampFormatInfo()),
                        new FieldInfo("proctime", new TimestampFormatInfo())),
                new FieldRelation(new FieldInfo("database_name", new StringFormatInfo()),
                        new FieldInfo("database_name", new StringFormatInfo())),
                new FieldRelation(new FieldInfo("collection_name", new StringFormatInfo()),
                        new FieldInfo("collection_name", new StringFormatInfo())),
                new FieldRelation(new FieldInfo("op_ts", new TimestampFormatInfo()),
                        new FieldInfo("op_ts", new TimestampFormatInfo())));
        CsvFormat csvFormat = new CsvFormat();
        csvFormat.setDisableQuoteCharacter(true);
        return new KafkaLoadNode("2", "kafka_output", fields, relations, null, null,
                "test", "localhost:9092",
                csvFormat, null,
                null, "_id");
    }
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