xây dựng job đồng bộ dữ liệu bằng java từ 2 bảng (Oracle DB) về 1 bảng log chung

static {
    try (InputStream input = DBUtil.class.getClassLoader().getResourceAsStream("sync_config.properties")) {
        Properties prop = new Properties();
        prop.load(input);
        dbUrl = prop.getProperty("db.url");
        dbUser = prop.getProperty("db.username");
        dbPass = prop.getProperty("db.password");
        Class.forName("oracle.jdbc.OracleDriver");
    } catch (Exception e) {
        e.printStackTrace();
    }
}

public static Connection getConnection() throws SQLException {
    return DriverManager.getConnection(dbUrl, dbUser, dbPass);
}

public void sync(String sourceTable) {
    System.out.println("Syncing from table: " + sourceTable + " at " + new Date());

    String query = "SELECT id, data, created_at FROM " + sourceTable + " WHERE created_at > (SELECT NVL(MAX(created_at), TO_DATE('2000-01-01','YYYY-MM-DD')) FROM log_table WHERE source_table = ?)";

    try (Connection conn = DBUtil.getConnection();
         PreparedStatement stmt = conn.prepareStatement(query)) {

        stmt.setString(1, sourceTable);
        ResultSet rs = stmt.executeQuery();

        int count = 0;

        while (rs.next()) {
            int id = rs.getInt("id");
            String data = rs.getString("data");
            Timestamp createdAt = rs.getTimestamp("created_at");

            insertLog(conn, sourceTable, id, data, createdAt);
            count++;
        }

        System.out.println("Synced " + count + " records from " + sourceTable);

    } catch (Exception e) {
        e.printStackTrace();
    }
}

private void insertLog(Connection conn, String sourceTable, int sourceId, String data, Timestamp createdAt) {
    String insertSQL = "INSERT /*+ IGNORE_ROW_ON_DUPKEY_INDEX(log_table, (source_table, source_id)) */ " +
                       "INTO log_table (id, source_table, source_id, data, created_at, synced_at) " +
                       "VALUES (log_seq.NEXTVAL, ?, ?, ?, ?, SYSTIMESTAMP)";
    try (PreparedStatement stmt = conn.prepareStatement(insertSQL)) {
        stmt.setString(1, sourceTable);
        stmt.setInt(2, sourceId);
        stmt.setString(3, data);
        stmt.setTimestamp(4, createdAt);
        stmt.executeUpdate();
    } catch (SQLIntegrityConstraintViolationException e) {
        // Trùng → bỏ qua
    } catch (SQLException e) {
        e.printStackTrace();
    }
}

        long interval = Long.parseLong(props.getProperty("sync.interval.seconds", "86400")); // Mặc định 24h

        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
        SyncService syncService = new SyncService();

        Runnable task = () -> {
            syncService.sync("SOURCE_A");
            syncService.sync("SOURCE_B");
        };

        scheduler.scheduleAtFixedRate(task, 0, interval, TimeUnit.SECONDS);

    } catch (Exception e) {
        e.printStackTrace();
    }
}