FlinkCDC同步oracle数据的时候，数据总是要个两三分钟才能同步过去

FlinkCDC同步oracle数据的时候，数据总是要个两三分钟才能同步过去，不管是插入的一条数据还是多条数据，间隔的时间也不固定。请问是设置有啥问题吗？代码如下：
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing();
env.getCheckpointConfig().setCheckpointInterval(2000);
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1500);
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");
}