请问Flink CDC OracleConnector支持从指定scn同步么?

是的，Flink CDC OracleConnector支持从指定的System Change Number (SCN)同步数据。你可以通过设置oracle.scn.startup.mode参数来指定起始SCN，从而实现从指定SCN开始同步数据。

以下是一个示例代码：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.hive.HiveCatalog;
import org.apache.flink.table.descriptors.SchemaDescriptor;
import org.apache.flink.table.descriptors.TableDescriptor;
import org.apache.flink.table.factories.TableFactory;
import org.apache.flink.table.factories.utils.FactoryUtil;
import org.apache.flink.table.types.DataTypes;

public class FlinkCDCExample {
    public static void main(String[] args) throws Exception {
        // 创建流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 创建表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 注册HiveCatalog
        HiveCatalog hiveCatalog = new HiveCatalog("default", "localhost:9083", "default");
        tableEnv.registerCatalog("hive", hiveCatalog);
        tableEnv.useCatalog("hive");

        // 定义源表和目标表的DDL
        String sourceDDL = "CREATE TABLE source_table (id INT, name STRING, age INT) WITH (...)";
        String sinkDDL = "CREATE TABLE sink_table (id INT, name STRING, age INT) WITH (...)";

        // 创建源表和目标表
        tableEnv.executeSql(sourceDDL);
        tableEnv.executeSql(sinkDDL);

        // 使用Flink CDC将源表的数据同步到目标表，并指定起始SCN为1234567890
        tableEnv.executeSql("""
            CREATE TABLE sync_table (
                ... -- 定义源表和目标表的字段类型和名称对应关系
            ) WITH (
                'connector' = 'oracle-cdc',
                'hostname' = 'localhost',
                'port' = '1521',
                'username' = 'your_username',
                'password' = 'your_password',
                'database' = 'your_database',
                'scnStartupMode' = 'specific-scn', -- 指定起始SCN模式为specific-scn
                'scn' = '1234567890' -- 指定起始SCN为1234567890
            ) AS SELECT * FROM source_table;
        """);
    }
}

在这个示例中，我们创建了一个名为sync_table的临时表，用于将源表source_table的数据同步到目标表sink_table。通过设置scnStartupMode参数为specific-scn，并指定起始SCN为1234567890，我们可以实现从指定SCN开始同步数据。