FlinkCDC同步Oracle的同一个库的多个表就会延迟很大

使用同一段代码，只是表名不同，目的是同步同一个库中的多个表。运行在yarn上的。当启动一个表的任务时，基本没什么延迟，并且一切正常，但是当再启动另一个表的同步任务时，就会特别慢，延迟四五分钟这样，哪怕新增一笔数据也是延迟这么久。有时候还会出现数据库连接失败的报错。具体代码如下：public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing();
env.getCheckpointConfig().setCheckpointInterval(30000);
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);
env.getCheckpointConfig().setCheckpointTimeout(60000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
env.setParallelism(1);

EnvironmentSettings Settings = EnvironmentSettings.newInstance()
        .inStreamingMode()
        .build();

StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, Settings);

String sourceDDL = "CREATE TABLE TS_XX_source (\n" +
                "ACCT_ID DECIMAL,\n" +
                " SEQ DECIMAL,\n" +
                " CC_TYPE DECIMAL,\n" +
                " CREDIT_LIMIT DECIMAL,\n" +
                " EFF_DATE TIMESTAMP(3),\n" +
                " EXP_DATE TIMESTAMP(3),\n" +
                " SP_ID DECIMAL,\n" +
                " primary key (ACCT_ID) not enforced\n" +
                ") WITH (\n" +
                " 'connector' = 'oracle-cdc',\n" +
                " 'hostname' = '******',\n" +
                " 'port' = '******',\n" +
                " 'username' = '******',\n" +
                " 'password' = '******',\n" +
                " 'database-name' = '******',\n" +
                " 'schema-name' = '******'," +
                " 'table-name' = '这里表名不同',\n" +
                " 'scan.startup.mode' = 'latest-offset'\n" +
                ")";

String sinkDDL = "CREATE TABLE TS_XX_sink (\n" +
                "ACCT_ID DECIMAL,\n" +
                " SEQ DECIMAL,\n" +
                " CC_TYPE DECIMAL,\n" +
                " CREDIT_LIMIT DECIMAL,\n" +
                " EFF_DATE TIMESTAMP(3),\n" +
                " EXP_DATE TIMESTAMP(3),\n" +
                " SP_ID DECIMAL,\n" +
                " primary key (ACCT_ID) not enforced\n" +
                ") WITH (\n" +
                " 'connector' = 'jdbc',\n" +
                " 'driver' = 'oracle.jdbc.driver.OracleDriver',\n" +
                " 'url' = '******',\n" +
                " 'username' = '****',\n" +
                " 'password' = '****',\n" +
                " 'table-name' = '****'\n" +
                ")";

String transformDmlSQL =  "insert into TS_XX_sink select * from TS_XX_source";

tableEnv.executeSql(sourceDDL).print();
tableEnv.executeSql(sinkDDL).print();
tableEnv.executeSql(transformDmlSQL).print();

env.execute("sync-cdc");

}