@Test
    public void testSqlServerExtract() throws Exception {
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .inStreamingMode()
                .build();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(10000);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
        Node sqlServerExtractNode = buildSqlServerExtractNode("1");
        Node kafkaLoadNode = buildKafkaNode("2");
        StreamInfo streamInfo = new StreamInfo("1L", Arrays.asList(sqlServerExtractNode, kafkaLoadNode),
                Collections.singletonList(buildNodeRelation(Collections.singletonList(sqlServerExtractNode),
                        Collections.singletonList(kafkaLoadNode))));
        GroupInfo groupInfo = new GroupInfo("1", Collections.singletonList(streamInfo));
        FlinkSqlParser parser = FlinkSqlParser.getInstance(tableEnv, groupInfo);
        Assert.assertTrue(parser.parse().tryExecute());
    }
Test extract data from sqlserver and load data to kafka.
inlong-sort/sort-core/src/test/java/org/apache/inlong/sort/parser/SqlServerNodeSqlParseTest.java