Add PriorityQueue/control backpressure for Scanner remote segments download logic like in Java Client

[fresh-borzoni]

Search before asking

• I searched in the issues and found nothing similar.

Description

in scanner.rs

    fn pending_remote_fetches(
        // ....
        // Download and process remote log segments
        let mut pos_in_log_segment = remote_fetch_info.first_start_pos;
        let mut current_fetch_offset = fetch_offset;
        for (i, segment) in remote_fetch_info.remote_log_segments.iter().enumerate() {
            if i > 0 {
                pos_in_log_segment = 0;
                current_fetch_offset = segment.start_offset;
            }

            // todo:
            // 1: control the max threads to download remote segment
            // 2: introduce priority queue to priority highest for earliest segment
            let download_future = remote_log_downloader
                .request_remote_log(&remote_fetch_info.remote_log_tablet_dir, segment);
//..

So we need to fix this and do something similar like in Java client

Willingness to contribute

• I'm willing to submit a PR!

[luoyuxia]

@fresh-borzoni Assigned to you, thanks

[zhaohaidao]

Hi, I'd like to know the progress of this issue. I encountered an OOM (Out of Memory) problem during long-term stability testing, and I understand that with this back pressure capability, OOM can be avoided.
@fresh-borzoni

[fresh-borzoni]

@zhaohaidao
I expect it to be done this week, currently working on it.

[luoyuxia]

Hi, I'd like to know the progress of this issue. I encountered an OOM (Out of Memory) problem during long-term stability testing, and I understand that with this back pressure capability, OOM can be avoided. @fresh-borzoni

I think this issue should resolve OOM for most of case, but It also remains a issue #157 which will read all bytes of a segment into memory. But it's an optimize, let's to see what will happen after we introduce back pressure

[zhaohaidao]

Hi, I'd like to know the progress of this issue. I encountered an OOM (Out of Memory) problem during long-term stability testing, and I understand that with this back pressure capability, OOM can be avoided. @fresh-borzoni

I think this issue should resolve OOM for most of case, but It also remains a issue #157 which will read all bytes of a segment into memory. But it's an optimize, let's to see what will happen after we introduce back pressure

Thanks for reminding. Let me assess the risks in the internal business scenario.

[zhaohaidao]

@zhaohaidao I expect it to be done this week, currently working on it.

Great. Looking forward it