@Override
    public Map<String, String> tableOptions() {
        Map<String, String> options = super.tableOptions();
        options.put(OracleConstant.CONNECTOR, OracleConstant.ORACLE_CDC);
        options.put(OracleConstant.HOSTNAME, hostname);
        options.put(OracleConstant.USERNAME, username);
        options.put(OracleConstant.PASSWORD, password);
        options.put(OracleConstant.DATABASE_NAME, database);
        options.put(OracleConstant.SCHEMA_NAME, schemaName);
        options.put(OracleConstant.TABLE_NAME, tableName);
        if (!options.containsKey(OracleConstant.TABLENAME_CASE_INSENSITIVE)) {
            // Set a default value:false to avoid retrieving unlisted tables in oracle 11g.
            // You can set null to replace it if your oracle version is more than oracle 11g.
            options.put(OracleConstant.TABLENAME_CASE_INSENSITIVE, "false");
        }
        if (!options.containsKey(OracleConstant.LOG_MINING_STRATEGY)) {
            // Set a default value:online_catalog to improve performance and reduce latency
            options.put(OracleConstant.LOG_MINING_STRATEGY, "online_catalog");
        }
        if (!options.containsKey(OracleConstant.LOG_MINING_CONTINUOUS_MINE)) {
            // Set a default value:true to improve performance and reduce latency
            options.put(OracleConstant.LOG_MINING_CONTINUOUS_MINE, "true");
        }
        if (port != null) {
            options.put(OracleConstant.PORT, port.toString());
        }
        if (scanStartupMode != null) {
            options.put(OracleConstant.SCAN_STARTUP_MODE, scanStartupMode.getValue());
        }
        return options;
    }
Oracle extract node for extract data from oracle(Currently support oracle 11,12,19)
inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/extract/OracleExtractNode.java