@Test
    public void testMySqlExtractNodeToDorisLoadNodeSqlParse() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(10000);
        env.disableOperatorChaining();
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .inStreamingMode()
                .build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
        Node inputNode = buildMysqlExtractNode();
        Node outputNode = buildDorisLoadNode();
        StreamInfo streamInfo = new StreamInfo("1", Arrays.asList(inputNode, outputNode),
                Collections.singletonList(buildNodeRelation(Collections.singletonList(inputNode),
                        Collections.singletonList(outputNode))));
        GroupInfo groupInfo = new GroupInfo("1", Collections.singletonList(streamInfo));
        FlinkSqlParser parser = FlinkSqlParser.getInstance(tableEnv, groupInfo);
        ParseResult result = parser.parse();
        try {
            Assert.assertTrue(result.tryExecute());
        } catch (Exception e) {
            log.error("An exception occurred: {}", e.getMessage());
        }
    }
Test flink sql task for extract is mysql {@link MySqlExtractNode} and load is doris {@link DorisLoadNode}
inlong-sort/sort-core/src/test/java/org/apache/inlong/sort/parser/MySqlExtractNodeToDorisLoadNodeTest.java