[client] Refactor RemoteLogDownloader to use chunked streaming instead of downloading whole segment file
- Dominant language
- Java
- Stars
- 2.1k
- Forks
- 625
- Avg merge
- 3d 14h
- Merged PRs (30d)
- 97
Description
### Search before asking
- [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar.
### Motivation
The current `RemoteLogDownloader` downloads the entire remote log segment file to local disk before any data can be consumed. This has several problems:
- Wasted download bandwidth: If the consumer only needs a portion of a segment (e.g., it unsubscribes, seeks, or closes mid-segment), the remaining bytes have already been downloaded and are simply discarded. For large segments (hundreds of MBs), this wastes significant remote storage bandwidth — which is especially costly for cloud object storage (OSS, S3) where egress is metered.
- High time-to-first-byte latency: The consumer must wait for the full segment to finish downloading before it can start processing any records.
- No flow control at chunk level: The only flow control is the prefetchSemaphore which limits the number of active segments, but does not limit how far ahead data is downloaded within a single segment. This means data continues to be eagerly downloaded even when the consumer is slow, further amplifying unnecessary bandwidth usage.
### Solution
Refactor `RemoteLogDownloader` to read remote log segments in configurable chunks (default 8 MB) instead of downloading the whole file at once:
- Introduce `RemoteSegmentChunkReader` to read fixed-size chunks from the remote FSDataInputStream, aligning on batch boundaries.
- Each chunk is appended to a local temp file and returned as a `FileLogRecords` slice (leveraging OS page cache, no JVM heap copy).
- Add two-level flow control:
1. prefetchSemaphore: limits the number of concurrently active segments (existing, default 4).
1. maxPrefetchChunks: limits the number of unconsumed chunks per segment (new, default 5). The downloader pauses when this limit is reached and resumes only when chunks are consumed.
- Use a continuationQueue to prioritize chunk continuation over new segment downloads.
- Chain chunk futures via nextChunkCallback so that LogFetcher automatically registers `RemotePendingFetch` for each subsequent chunk.
- New config options: `client.scanner.remote-log.chunk-size` (default 8 MB), `client.scanner.remote-log.max-prefetch-chunks` (default 5).
### Anything else?
_No response_
### Willingness to contribute
- [x] I'm willing to submit a PR!
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.