flink通过JDBC连接Oracle

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

DataStreamSource<Row> input = env.createInput(JDBCInputFormat.buildJDBCInputFormat()
        .setDrivername("oracle.jdbc.driver.OracleDriver")
        .setDBUrl("jdbc:oracle:thin:@192.168.***.***:1521/orcl")
        .setUsername("FCDT")
        .setPassword("FCDT0606")
        .setQuery("select id from FCDT.X_NEWS_TCRNW0002 ")
        .setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.BIG_DEC_TYPE_INFO))
        .finish());
Table table = tableEnv.fromChangelogStream(input);
table.execute().print();